Патчим процессы в Linux на лету при помощи GDB
Техники перехвата функций в Linux хорошо известны и описаны в интернете. Наиболее простой метод заключается в написании динамической библиотеки с «функциями-клонами» и использовании механизма LD_PRELOAD для переопределения таблицы импорта на этапе загрузки процесса.Недостаток LD_PRELOAD в том что необходимо контролировать запуск процесса. Для перехвата функций в уже работающем процессе или функций отсутствующих в таблице импорта можно использовать «сплайсинг» — запись команды перехода на перехватчик в начало перехватываемой функции.
Также известно, что в Python имеется модуль ctypes позволяющий взаимодействовать с данными и функциями языка Си (т.е. большим числом динамических библиотек имеющих Си интерфейс). Таким образом ничто не мешает перехватить функцию процесса и направить её в Python метод обёрнутый в С-callback с помощью ctypes.Для перехвата управления и загрузки кода в целевой процесс удобно использовать отладчик GDB, который поддерживает написание модулей расширения на языке Python (https://sourceware.org/gdb/current/onlinedocs/gdb/Python-API.html).
Нюансы Код примера приведен полностью в конце статьи и состоит из двух файлов: pyinject.py — расширение GDB hook.py — модуль с функциями перехватчиками Со стороны GDB код удобно оформить в виде пользовательской команды. Новую команду можно создать, наследуя от класса gdb.Command. При использовании команды в GDB будет вызываться метод invoke (argument, from_tty).Также можно создавать пользовательские параметры наследуя от gdb.Parameter. В примере статьи он используется для задания имени файла с функциями перехвата.
Подключение к работающему процессу PID и загрузку модуля удобно делать сразу при запуске GDB
gdb -ex 'attach PID' -ex 'source pyinject.py' -ex 'set hookfile hook.py' Поле этого отлаживаемый процесс остановлен и запущена интерактивная командная строка GDB, в которой будет доступна новая команда «pyinject». Перехват можно условно разделить на три этапа: Инжектирование интерпретатора Python в адресное пространство целевого процесса Сбор информации о перехватываемой функции Собственно перехват Пункты 1 и 2 проще делать на стороне отладчика, пункт 3 уже внутри целевого процесса.Инжектирование интерпретатора Python Большая часть Python интерфейса GDB предназначена для расширения отладочных возможностей. Для всего остального есть gdb.execute (command, from_tty, to_string), которая позволяет выполнить произвольную команду GDB и получить её вывод в виде строки.Например: out = gdb.execute («info registers», False, True) Также полезна gdb.parse_end_eval (expression), вычисляющая выражение и возвращающая результат в виде gdb.Value.Первым делом необходимо загрузить библиотеку Python в адресное пространство целевого процесса. Для этого необходимо вызвать dlopen в контексте целевого процесса.Можно использовать команду call в gdb.execute, либо gdb.parse_and_eval:
# pyinject.py gdb.execute ('call dlopen («libpython2.7.so», %d)' % RTLD_LAZY) assert long (gdb.history (0)) handle = gdb.parse_and_eval ('dlopen («libpython2.7.so», %d)' % RTLD_LAZY) assert long (handle) После этого можно инициализировать интерпретатор # pyinject.py gdb.execute ('call PyEval_InitThreads ()') gdb.execute ('call Py_Initialize ()') Первый вызов создает GIL (global interpreter lock), второй подготавливает Python C-API к использованию.И загрузить модуль с функциями перехвата
# pyinject.py fp = gdb.parse_and_eval ('fopen («hook.py», «r»)') assert long (fp) != 0 pyret = gdb.parse_and_eval ('PyRun_AnyFileEx (%u, «hook.py», 1)' % fp) PyRun_AnyFileEx выполняет код из файла в контексте модуля __main__. Нюансы Вышеописанное будет работать только если целевой процесс не использует Python (как основной или скриптовый язык). Если это не так, то всё серьёзно усложняется. Основная проблема в том что в процессе остановленном для отладки в случайном месте нельзя использовать никакие функции Python C-API (кроме может быть Py_AddPendingCall).
Модуль hook.py Модуль hook.py содержит функции перехватчики и класс Hook выполняющий собственно перехват.Функции перехватчики обозначаются при помощи декоратора. Например для функции open стандартной библиотеки напечатаем её аргументы и вернем результат вызова оригинальной функции, хранящейся в поле orig # hook.py @hook (symbol='open', ctype=CFUNCTYPE (c_int, c_char_p, c_int)) def python_open (fname, oflag): print «open:», fname, oflag return python_open.orig (fname, oflag) Декоратор @hook принимает два параметра: symbol — имя перехватываемого символа (предполагается что символ доступен в GDB из таблиц импорта или отладочной информации, но ничто не мешает перехватывать функции по адресам вместо символов) ctype — класс ctypes задающий тип функции Декоратор регистрирует функцию в классе Hook и возвращает не изменяя. # hook.py def hook (symbol, ctype): def deco (func): Hook.register (symbol, ctype, func) return func return deco Метод register создает экземпляр класса и сохраняет его в словаре all_hooks. Таким образом после выполнения файла, благодаря декораторам в Hook.all_hooks будет вся информация о доступных функциях перехватчиках. # hook.py class Hook (object): all_hooks = {} @staticmethod def register (symbol, *args): Hook.all_hooks[symbol] = Hook (symbol, *args) Чтобы осуществить перехват со стороны GDB вызовом одной функции, удобно определить статический метод в классе Hook, ответственный за перехват # hook.py class Hook (object): @staticmethod def hook (symbol, *args): h = Hook.all_hooks[symbol] if h.active: return h.install (*args) В *args здесь передается дополнительная информация о перехватываемой функции. Какая именно зависит от метода перехвата.Методы перехвата «сплайсингом» Сплайсинг глобально делится на два подвида по способу вызова оригинальной функции.В simple hook вызов оригинальной функции состоит из нескольких шагов:
начало оригинальной функции восстанавливается из сохраненной копии производится вызов начало снова затирается инструкцией перехода на перехватчик Нюансы Недостаток очевиден, в многопоточной программе нельзя гарантировать, что другой поток не вызовет функцию во время перезаписи её начала. Частично это лечится остановкой других потоков на время вызова оригинальной функции. Но во-первых нет стандартного способа этого достичь, во-вторых можно словить deadlock если неудачно вызвать функцию типа malloc
В trampoline hook начало оригинальной функции копируется в новое место и после него записывается переход в тело оригинальной функции. В этом варианте оригинальная функция всегда доступна по новому адресу.Trampoline hook работает в многопоточных программах, но гораздо сложнее в установке. Необходимо перезаписывать целое число инструкций, для чего обычно используется дизассемблер. Приход архитектуры x86_64 добавил еще больше проблем из-за повсеместного распространения адресации памяти относительно регистра %rip (адрес текущей команды).
Нюансы
Посмотрим на начало функции open в GDB:
0×7f6cc8aa83e0
определить размер команд в начале функции выделить память не дальше чем в 2ГБ от целевого адреса команды cmpl (смещение 0×2d33ed (%rip) знаковое 32-битное) скопировать начало в новое место и пропатчить доступ к памяти относительно %rip в cmpl В довершение картины, команда перехода должна быть короче 9 байт, т.к. это функция с двумя точками входа и по адресу 0×7f6cc8aa83e9 уже находится __open_nocancel. Это значит, что наш трамплин должен быть не дальше чем в 2ГБ от начала open для возможности 32-битного перехода (все 64-битные переходы длиннее 9 байт). В принципе, имея всю мощь GDB за спиной (gdb.execute ()), ничто не мешает корректно реализовать trampoline hook, но для простоты примера в этой статье будет использоваться simple hook.В simple hook единственное ограничение это длина инструкции перехода.Вариантов два (основных):
В статье используется второй метод # hook.py class Hook (object): @staticmethod def get_indlongjmp (srcaddr, proxyaddr): s = struct.pack ('=BBl', 0xff, 0×25, proxyaddr — srcaddr — 6) return map (ord, s) get_indlongjmp возвращает код для прыжка с адреса srcaddr на адрес сохраненный в QWORD по адресу proxyaddrТеперь можно наконец написать недостающие методы класса Hook. Метод install получает адрес оригинальной функции address и адрес вспомогательной зоны proxyaddr. После чего переписывает начало функции (предварительно сохранив его в self.code) переходом на перехватчик
# hook.py def install (self, address, proxyaddr): self.address = address self.proxyaddr = proxyaddr proxymemory = (c_void_p * 1).from_address (self.proxyaddr) proxymemory[0] = Hook.cast_to_void_p (self.cfunc) self.jmp = self.get_indlongjmp (self.address, self.proxyaddr) self.memory = (c_ubyte * len (self.jmp)).from_address (self.address) self.code = list (self.memory) self.patchmem (self.jmp) self.pyfunc.orig = self.origfunc () self.active = True patchmem перезаписывает начало оригинальной функции данными из src # hook.py def patchmem (self, src): for i in range (len (src)): self.memory[i] = src[i] origfunc оборачивает вызов функции в код снимающий и устанавливающий переход на перехватчик. # hook.py def origfunc (self): ofunc = self.ctype (self.address) def wrap (*args): self.patchmem (self.code) val = ofunc (*args) self.patchmem (self.jmp) return val return wrap Последние штрихи Python загружен в адресное пространство, файл hook.py загружен в Python. Осталось вызвать Hook.hook (symbol, address, proxyaddr) cо стороны Python модуля GDB.Находим адрес функции «open»
line = gdb.execute ('info address %s' % «open» False, True) m = re.match (r'.*?(0x[0–9a-f]+)', line) addr = int (m.group (1), 16) Нюансы В общем случае, перед тем как бежать переписывать код остановленного процесса надо убедиться что он не остановлен посередине этого кода (или собирается вернуться в него). Сделать это проще всего, отпарсив вывод gdb.execute («thread apply all backtrace»)
Выделяем память поблизости от addr prot = PROT_READ | PROT_WRITE | PROT_EXEC flags = MAP_PRIVATE | MAP_ANONYMOUS maddr = gdb.parse_and_eval ('(void*)mmap (0x%x, %d, %d, %d, -1, 0)\n' % (addr | 0×7FFFFFFF, 4096, prot, flags)) maddr = (long (maddr) & 0×00000000FFFFFFFF) | (addr & 0xFFFFFFFF00000000) Нюансы Последняя строка это обход бага в GDB, который съедает старшие биты результата. Аргумент (addr | 0×7FFFFFFF) использует недокументированное свойство mmap выдавать память с адресом меньше занятого желаемого.Без трюков по-правильному чуть длиннее: надо отпарсить вывод gdb.execute ('info proc mappings', False, True), найти ближайшую к addr дырку в адресном пространстве и вывать mmap с MAP_FIXED. Ну и естественно не обязательно выделять по целой странице памяти для каждой перехваченой функции.
Разрешаем перезапись оригинальной функции (иначе SIGSEGV) gdb.parse_and_eval ('mprotect (0x%x, %u, %d)' % (addr & -0×1000, 4096×2, prot)) Вызываем Hook.hook через PyRun_SimpleString pyret = gdb.parse_and_eval ('PyRun_SimpleString («Hook.hook (\\«open\\», 0x%x, 0x%x)»)' % (addr, maddr)) Готово! Теперь вызов «open» в целевом процессе будет перехвачен и направлен в python_open из hook.py.Файлы примеров Полные файлы примеров (с чуть большим количеством проверок, но без учета многих нюансов)pyinject.py # pyinject.py import re import os
RTLD_LAZY = 1 PROT_READ = 0×1 PROT_WRITE = 0×2 PROT_EXEC = 0×4 MAP_PRIVATE = 0×2 MAP_FIXED = 0×10 MAP_ANONYMOUS = 0×20 LIBPYTHON = 'libpython2.7.so'
class ParamHookfile (gdb.Parameter): instance = None def __init__(self, default=''): super (ParamHookfile, self).__init__(«hookfile», gdb.COMMAND_NONE, gdb.PARAM_FILENAME) self.value = default ParamHookfile.instance = self
def get_set_string (self): return self.value
def get_show_string (self, svalue): return svalue
class CmdHook (gdb.Command): instance = None def __init__(self): super (CmdHook, self).__init__(«pyinject», gdb.COMMAND_NONE) self.initialized = False CmdHook.instance = self
def complete (self, text, word): matching = [s[4:] for s in dir (self) if s.startswith ('cmd_') and s[4:].startswith (text)] return matching
def invoke (self, subcmd, from_tty): self.dont_repeat () if subcmd.startswith («hook»): self.cmd_hook (*gdb.string_to_argv (subcmd)) elif subcmd.startswith («unhook»): self.cmd_unhook (*gdb.string_to_argv (subcmd)) else: gdb.write ('unknown sub-command »%s»' % subcmd)
def cmd_hook (self, *args): self.initialize () if not self.initialized: return
pyret = gdb.parse_and_eval ('PyRun_SimpleString («print Hook»)')
if long (pyret) != 0:
hookfile = ParamHookfile.instance.value
if not os.path.exists (hookfile):
gdb.write ('Use «set hookfile
for symbol in args: try: line = gdb.execute ('info address %s' % symbol, False, True) m = re.match (r'.*?(0x[0–9a-f]+)', line) if m: addr = int (m.group (1), 16) except gdb.error: continue prot = PROT_READ | PROT_WRITE | PROT_EXEC flags = MAP_PRIVATE | MAP_ANONYMOUS # | MAP_FIXED maddr = gdb.parse_and_eval ('(void*)mmap (0x%x, %d, %d, %d, -1, 0)\n' % (addr | 0×7FFFFFFF, 4096, prot, flags)) maddr = (long (maddr) & 0×00000000FFFFFFFF) | (addr & 0xFFFFFFFF00000000) gdb.write («mmap = 0x%x\n» % maddr) if maddr == 0: continue gdb.parse_and_eval ('mprotect (0x%x, %u, %d)' % (addr & -0×1000, 4096×2, prot)) pyret = gdb.parse_and_eval ('PyRun_SimpleString («Hook.hook (\\»%s\\», 0x%x, 0x%x)»)' % (symbol, addr, maddr)) if long (pyret) == 0: gdb.write ('hook »%s» OK\n' % symbol)
def cmd_unhook (self, *args): for symbol in args: pyret = gdb.parse_and_eval ('PyRun_SimpleString («Hook.unhook (\\»%s\\»)»)' % (symbol)) if long (pyret) == 0: gdb.write ('unhook »%s» OK\n' % symbol)
def initialize (self): if self.initialized: return handle = gdb.parse_and_eval ('dlopen (»%s», %d)' % (LIBPYTHON, RTLD_LAZY)) if not long (handle): gdb.write ('Cannot load library %s\n' % LIBPYTHON) return if not long (gdb.parse_and_eval ('Py_IsInitialized ()')): gdb.execute ('call PyEval_InitThreads ()') gdb.execute ('call Py_Initialize ()') self.initialized = True
if __name__ == '__main__': ParamHookfile () CmdHook () hook.py # hook.py import struct from ctypes import (CFUNCTYPE, POINTER, c_ubyte, c_int, c_char_p, c_void_p)
class Hook (object): all_hooks = {}
@staticmethod def cast_to_void_p (pointer): return CFUNCTYPE (c_void_p, c_void_p)(lambda x: x)(pointer)
@staticmethod def register (symbol, *args): Hook.all_hooks[symbol] = Hook (symbol, *args)
def __init__(self, symbol, ctype, pyfunc): self.symbol = symbol self.ctype = ctype self.pyfunc = pyfunc self.cfunc = self.ctype (self.pyfunc) self.address = 0 self.proxyaddr = 0 self.jmp = None self.memory = None self.code = None self.active = False
def install (self, address, proxyaddr): print «install:», hex (address) self.address = address self.proxyaddr = proxyaddr proxymemory = (c_void_p * 1).from_address (self.proxyaddr) proxymemory[0] = Hook.cast_to_void_p (self.cfunc) self.jmp = self.get_indlongjmp (self.address, self.proxyaddr) self.memory = (c_ubyte * len (self.jmp)).from_address (self.address) self.code = list (self.memory) self.patchmem (self.jmp) self.pyfunc.orig = self.origfunc () self.active = True
def uninstall (self): self.patchmem (self.code) self.active = False
def origfunc (self): ofunc = self.ctype (self.address) def wrap (*args): self.patchmem (self.code) val = ofunc (*args) self.patchmem (self.jmp) return val return wrap
def patchmem (self, src): for i in range (len (src)): self.memory[i] = src[i]
@staticmethod def get_indlongjmp (srcaddr, proxyaddr): # 64-bit indirect absolute jump (6 + 8 bytes) # ff 25 off32 jmpq *off32(%rip) try: s = struct.pack ('=BBl', 0xff, 0×25, proxyaddr — srcaddr — 6) return map (ord, s) except: print hex (proxyaddr), hex (srcaddr), hex (proxyaddr — srcaddr — 6) raise
@staticmethod def hook (symbol, address, proxyaddr): h = Hook.all_hooks[symbol] if h.active: return h.install (address, proxyaddr)
@staticmethod def unhook (symbol): h = Hook.all_hooks[symbol] if not h.active: return h.uninstall ()
def hook (symbol, ctype): def deco (func): Hook.register (symbol, ctype, func) return func return deco
#int open (const char *__file, int __oflag, …) @hook (symbol='open', ctype=CFUNCTYPE (c_int, c_char_p, c_int)) def python_open (fname, oflag): print «open:», fname, oflag return python_open.orig (fname, oflag) Запуск примера (лучше с абсолютными путями) gdb -ex 'attach PID' -ex 'source /path/pyinject.py' -ex 'set hookfile /path/hook.py' (gdb) pyinject hook open (gdb) continue
