Пишем низкоуровневый отладчик под Linux на Python

Существуют отличные отладчики вроде GDB и LLDB. И хотя их можно настраивать с помощью скриптов, порой хочется иметь больше контроля над работой отладчика. В этой серии статей мы попробуем создать свой отладчик с помощью библиотек python-ptrace, pyelftools и distorm3.

Исходники доступны на GitHub. Всё написано и скомпилировано на Linux x86_64.

Прим.перев. В этой статье используется устаревшая версия Python 2.

Подготовка

Чтобы избежать проблем с правами доступа, мы будем запускать отлаживаемый процесс как дочерний:

import ptrace.debugger
shell_command = ["./a.out"]
child_proc = subprocess.Popen(shell_command)
pid = child_proc.pid
debugger = ptrace.debugger.PtraceDebugger()
process = debugger.addProcess(pid, False)

Здесь используется системный вызов ptrace для присоединения к дочернему процессу и его остановки. Теперь process содержит много удобных методов. Идея была заимствована из примера в документации python-ptrace.

Считываем значения

Начнём с простого. Получаем регистры:

>>> regs = process.getregs()
>>> registers = {k: getattr(regs, k) for k in dir(regs) if not k.startswith('_')}
>>> registers
{'cs': 51L,
 'ds': 0L,
 'eflags': 519L,
 [...]
 'rax': 0L,
 'rbp': 140733962602848L,
 'rbx': 3L,
 'rcx': 139901135742274L,
 'rdi': 3L,
 'rdx': 140733962602656L,
 'rip': 139901135742280L,
 'rsi': 140733962602656L,
 'rsp': 140733962602520L,
 'ss': 43L}

Считываем байты из памяти:

>>> import binascii
>>> binascii.hexlify(process.readBytes(registers['rsp'], 8))
'70987e453d7f0000'

Ассемблерный REPL

Теперь нам нужно научиться запускать ассемблерные инструкции по одной за раз. Давайте соберём нужные составляющие.

Одиночный шаг:

>>> process.getreg('rip')
140187902313503L
>>> process.singleStep()
>>> process.getreg('rip')
140187902313507L

rip — это указатель на инструкцию. Префикс r обозначает длину в 64 бита. Как видите, он действительно смещается вперёд, когда мы делаем шаг.

Продолжаем до тех пор, пока дочерний процесс не получит сигнал (в нашем случае SIGTRAP). Это может привести к ошибке, если процесс завершится или будет получен другой сигнал:

>>> import signal
>>> process.waitSignals(signal.SIGTRAP)
ProcessSignal('Signal SIGTRAP',)

process.singleStep() неблокирующий, поэтому для удобства мы добавим блокирующую версию:

def step():
    process.singleStep()
    process.waitSignals(signal.SIGTRAP)

Так делать не стоит, но пусть process пока побудет глобальной переменной.

Пишем в память. В ассемблере выполнение ассемблерной инструкции int3 приводит к тому, что процессу отправляется сигнал SIGTRAP. Её можно записать в виде одного байта 0xCC:

>>> process.writeBytes(process.getreg('rip'), chr(0xCC))
>>> process.cont()
>>> process.waitSignals(signal.SIGTRAP)
ProcessSignal('Signal SIGTRAP',)

Также мы можем сравнить регистр rip до и после, чтобы проверить, что значение увеличилось ровно на 1.

Устанавливаем регистр:

>>> process.setreg('rax', 0)

Теперь у нас есть всё, что нужно, для запуска одной инструкции, переданной в виде байтов:

def run_asm(instr):
    old_rip = process.getreg('rip')
    old_values = process.readBytes(old_rip, len(instr))
    process.writeBytes(old_rip, instr)
    step()
    # Отматываем rip, если инструкция его не изменила.
    if process.getreg('rip') == old_rip + len(instr):
        process.setreg('rip', old_rip)
    process.writeBytes(old_rip, old_values)

Здесь мы перезаписываем байты перед указателем на инструкцию с помощью instr, делаем шаг и возвращаем перезаписанные байты и позицию указателя инструкции. Последнюю часть делаем, только если указатель на инструкцию не изменялся (как в случае с jump или call).

При помощи таблицы преобразования ассемблерных инструкций в байты мы можем поместить это в цикл и сделать REPL.

Вызов функции, первая попытка

Что делать, если мы хотим вызвать ассемблерную функцию и приостановить выполнение после возвращения из неё?

Напишем для этого func_call(func_addr) (запустите её пошагово, чтобы посмотреть на промежуточные состояния). Сначала сохраним часть текущего состояния:

def func_call(func_addr):
    old_rip = process.getreg('rip')
    old_regs = process.getregs()
    old_values = process.readBytes(old_rip, 6)

Мы могли бы просто использовать run_asm с инструкцией call. Это байт 0xE8, за которым следуют 5 байт little endian, описывающих разницу между текущим и целевым значениями rip.

Чтобы приостановить дочерний процесс после вызова, мы можем записать int3 (байт 0xCC) после инструкций вызова:

    diff = func_addr - (old_rip + 5)
    new_values = chr(0xE8) + struct.pack('i', diff) + chr(0xCC)
    process.writeBytes(old_rip, new_values)
    step()

Мы можем перепроверить, что вызов был совершён:

    new_rip = process.getreg('rip')
    assert(new_rip == func_addr)

Теперь пусть процесс работает, пока не будет получен сигнал SIGTRAP (желательно тот, что мы установили):

    process.cont()
    process.waitSignals(signal.SIGTRAP)

А теперь восстановим перезаписанные байты и значения регистра. В некоторых ситуациях они нам могут пригодиться:

    process.writeBytes(old_rip, old_values)
    process.setregs(old_regs)

Получаем адрес функции

Давайте попробуем вызвать скомпилированные Си-функции, но пока без аргументов и возвращаемого значения. Для этого нам всего лишь нужно найти адрес функции. Мы можем его получить из заголовка с помощью pyelftools:

from elftools.elf.elffile import ELFFile
from elftools.elf.sections import SymbolTableSection

def variables(filename="a.out"):
    f = ELFFile(open(filename))
    symb_sections = [section for section in f.iter_sections()
                     if isinstance(section, SymbolTableSection)]
    variables = {symb.name: symb['st_value'] for section in symb_sections
                 for symb in section.iter_symbols()}
    return variables

А теперь сам вызов:

>>> c_variables = variables("a.out")
>>> func_call(c_variables['some_func_name'])

Вообще, этот метод получает не только функции, но и, наверное, все статические переменные. Для библиотек общего пользования мы можем вызвать variables с полным путём к .so-файлу соответствующей библиотеки.

Тем не менее всегда это работать не будет, поскольку фактический регион используемой памяти не всегда начинается с 0 и нам нужно добавлять начало этого региона в качестве смещения.

Пока что мы можем это сделать следующим образом. С регионами памяти и /proc/pid/maps разберёмся чуть позже:

>>> line1 = open("/proc/%s/maps" % pid).readline()
>>> _start = int(line1.split("-")[0], 16)
>>> start = _start if _start != 0x400000 else 0
>>> func_call(start + c_variables['some_func_name'])

Ставим точки останова

Теперь у нас есть адреса функций и мы можем поставить точку останова, просто написав int3 (байт 0xCC) в начале функции:

def set_breakpoint(addr):
    old = process.readBytes(addr, 1)
    process.writeBytes(addr, chr(0xCC))
    return old

И восстановить перезаписанное значение после прохождения точки останова:

def restore_breakpoint(old):
    rip = process.getreg('rip')
    process.setreg('rip', rip - 1)
    addr = rip - 1
    process.writeBytes(addr, old)

Эти функции можно использовать следующим образом:

>>> old = set_breakpoint(start + variables['my_func'])
>>> process.waitSignals(signal.SIGTRAP)
>>> restore_breakpoint(old)

Вызов функции, вторая попытка

В общем и целом первый подход работает на удивление хорошо, хотя есть некоторые проблемы.

Слишком большое расстояние вызова. call (0xE8) принимает в качестве аргумента только 5 байт, однако для описания адреса (diff) может потребоваться 8 байт. Мы можем либо подождать, пока не окажемся в пределах функции, которую хотим вызвать (это работает только в том случае, если нам не нужно вызывать функцию сразу же), либо поместить целевой адрес в регистр, например, rax, и воспользоваться инструкцией call rax (байты FF D0).

Перезаписанные байты. Так как мы перезаписываем 7 байт (6 для call, один для int) и восстанавливаем их только после возвращения из функции, то в случае попытки их чтения из другого места можно получить неожиданные значения. Например, если мы совершили вызов внутри тела функции и выполнение программы снова доходит до old_rip.

В теории мы могли бы восстановить 6 из 7 байт после одного шага, оставив только 0xCC. Однако это не решает проблему, а только уменьшает её размер.

Ещё мы могли бы вручную создать стековый кадр.

Вместо этого мы зарезервируем новый участок памяти и запишем наши инструкции туда.

Выделяем память

Мы можем использовать системный вызов mmap() (номер вызова 9) для резервирования памяти. Ему требуются некоторые магические константы, часть которых можно найти в ptrace.syscall:

import ptrace.syscall
MMAP_PROT_BITMASK = {k: v for v, k in ptrace.syscall.posix_arg.MMAP_PROT_BITMASK}
MMAP_PROT_BITMASK['PROT_ALL'] = MMAP_PROT_BITMASK['PROT_READ']\
                              | MMAP_PROT_BITMASK['PROT_WRITE']\
              | MMAP_PROT_BITMASK['PROT_EXEC']
MAP_PRIVATE = 0x02
MAP_ANONYMOUS = 0x20
syscalls = {k: v for v, k in ptrace.syscall.linux_syscall64.SYSCALL_NAMES.items()}

С помощью следующей функции мы можем вызвать mmap. Здесь syscall представлен байтами 0F 05:

def reserve_memory(size):
    old_regs = process.getregs()
    regs = {'rax': syscalls['mmap'], 'rdi': 0, 'rsi': size,
            'rdx': MMAP_PROT_BITMASK['PROT_ALL'],
            'r10': MAP_PRIVATE | MAP_ANONYMOUS,
            'r8': -1, 'r9': 0}
    for reg, value in regs.items():
        process.setreg(reg, value)
    run_asm(chr(0x0f) + chr(0x05))
    result = process.getreg('rax')
    process.setregs(old_regs)
    return result

Данная стратегия была позаимствована из этого примера. Для справки, вот константы:

syscalls['mmap'] = 9
MMAP_PROT_BITMASK['PROT_ALL'] = 7
MAP_PRIVATE | MAP_ANONYMOUS = 34

Адрес зарезервированной памяти находится в rax после вызова, поэтому мы его извлекаем и возвращаем.

Это позволяет нам изменить вызов функции, сделав его немного безопаснее:

def safe_func_call(func_addr):
    old_rip = process.getreg('rip')
    old_regs = process.getregs()
    tmp_addr = reserve_memory(6)
    process.setreg('rip', tmp_addr)
    # call rax
    process.setreg('rax', func_addr)
    new_values = chr(0xff) + chr(0xd0) + chr(0xcc)
    process.writeBytes(tmp_addr, new_values)
    step()

    new_rip = process.getreg('rip')
    assert(new_rip == func_addr)
    process.cont()
    process.waitSignals(signal.SIGTRAP)
    process.setregs(old_regs)

Тем не менее, в этой функции по-прежнему могут возникать ошибки сегментации.

Получаем следующие инструкции

Добавим в наш отладчик функцию, которая говорит нам, какие следующие инструкции. Для этого нам понадобится дизассемблер distorm3, который можно установить с помощью pip.

Воспользуемся методом PtraceProcess.disassemble для получения итератора по следующим десяти инструкциям:

def look(addr=None):
    print("ip:", hex(process.getreg('rip')))
    for i, instr in enumerate(process.disassemble(start=addr)):
        hexa = instr.hexa
        hexa = ' '.join(hexa[i:i+2] for i in range(0, len(hexa), 2))
        print(str(i).ljust(4), hexa.ljust(24), instr.text.lower())

Запуск этой функции даст примерно следующий результат:

>>> look()
ip: 0x555c9860810dL
0    48 89 c2                 mov rdx, rax
1    48 8d 05 79 0f 20 00     lea rax, [rip+0x200f79]
2    48 89 10                 mov [rax], rdx
3    48 8d 05 6f 0f 20 00     lea rax, [rip+0x200f6f]
4    48 8b 00                 mov rax, [rax]
5    48 89 c6                 mov rsi, rax
6    48 8d 3d af 02 00 00     lea rdi, [rip+0x2af]
7    b8 00 00 00 00           mov eax, 0x0
8    e8 d8 fa ff ff           call 0x555c98607c10
9    48 8d 05 51 0f 20 00     lea rax, [rip+0x200f51]

Метод PtraceProcess.dumpCode работает похожим образом, но с другим форматированием.

Итог

На этом пока всё. В следующей статье мы разберёмся с чтением/записью Си-переменных, запуском одиночных Си-команд, библиотеками общего пользования, динамической загрузкой и картами памяти (/proc/pid/maps).

Перевод статьи «Making a low level (Linux) debugger»