frida hook工具使用——用于os api注入分析还是不错的

准备：

pip install frida

pip install frida-tools

开始：

1、创建child-gating1.py

import os
import threading

from frida_tools.application import Reactor

import frida

import argparse


class Application:
    def __init__(self, log_location):
        self.fpath = ""
        self.log = log_location
        
        self._stop_requested = threading.Event()
        self._reactor = Reactor(run_until_return=lambda reactor: self._stop_requested.wait())

        self._device = frida.get_local_device()
        self._sessions = set()

        self._device.on("child-added", lambda child: self._reactor.schedule(lambda: self._on_child_added(child)))
        self._device.on("child-removed", lambda child: self._reactor.schedule(lambda: self._on_child_removed(child)))
        self._device.on("output", lambda pid, fd, data: self._reactor.schedule(lambda: self._on_output(pid, fd, data)))

    def run(self, fp):
        self.fpath = fp
        self._reactor.schedule(lambda: self._start())
        self._reactor.run()

    def _start(self):
        """
        argv = ["/bin/sh", "-c", "cat /etc/hosts"]
        env = {
            "BADGER": "badger-badger-badger",
            "SNAKE": "mushroom-mushroom",
        }
        """
        #f = "C:\\Users\\Administrator\\Desktop\\测试用例\\TN0940\\notepad.exe"
        #f = "C:\\Users\\Administrator\\Desktop\\fritest\\TN0940_1.exe"
        print("✔ spawn(f={})".format(self.fpath))
        pid = self._device.spawn(self.fpath)
        self._instrument(pid)

    def _stop_if_idle(self):
        if len(self._sessions) == 0:
            self._stop_requested.set()

    def _instrument(self, pid):
        print(f"✔ attach(pid={pid})")
        session = self._device.attach(pid)
        session.on("detached", lambda reason: self._reactor.schedule(lambda: self._on_detached(pid, session, reason)))
        print("✔ enable_child_gating()")
        session.enable_child_gating()
        print("✔ create_script()")
        script = session.create_script(
            """
Interceptor.attach(Module.findExportByName("kernel32.dll", "CreateFileW"), {
  onEnter: function (args) {
    //var fileNamePtr = args[0];
    //var fileName = Memory.readUtf16String(fileNamePtr);
    //send(fileName)
    send({
      type: 'CreateFileW',
      path: Memory.readUtf16String(args[0])
    });
  }
});
"""
        )
        script.on("message", lambda message, data: self._reactor.schedule(lambda: self._on_message(pid, message)))
        print("✔ load()")
        script.load()
        print(f"✔ resume(pid={pid})")
        self._device.resume(pid)
        self._sessions.add(session)

    def _on_child_added(self, child):
        print(f"⚡ child_added: {child}")
        self._instrument(child.pid)

    def _on_child_removed(self, child):
        print(f"⚡ child_removed: {child}")

    def _on_output(self, pid, fd, data):
        print(f"⚡ output: pid={pid}, fd={fd}, data={repr(data)}")

    def _on_detached(self, pid, session, reason):
        print(f"⚡ detached: pid={pid}, reason='{reason}'")
        self._sessions.remove(session)
        self._reactor.schedule(self._stop_if_idle, delay=0.5)

    def _on_message(self, pid, message):
        print(f"⚡ message: pid={pid}, payload={message['payload']}")
        
        with open(self.log, 'a') as f:
            f.write(str(message.get('payload'))+'\n')
        

def log_loc(fpath, log_dir):
    malware_name = os.path.basename(fpath)[:-4]
    return os.path.join(log_dir, malware_name+".log")
    
  
parser = argparse.ArgumentParser(description='Frida demo.')
parser.add_argument("-f", "--file", help="target file to run", required=True)
parser.add_argument("-l", "--log", help="log location", required=True)
args, unknown = parser.parse_known_args()

app = Application(log_loc(fpath=args.file, log_dir=args.log))
app.run(args.file)

# sample run:
# python .\child-gating1.py -f ‪"C:\\Users\\l00379637\\source\\repos\\test_file_write\\Debug\\test_file_write.exe" -l read_file.log  

vs里创建一个测试exe，test_file_write：

#include <windows.h>
#include <stdio.h>

int main()
{
	HANDLE h = CreateFile(
		"bar.txt",
		GENERIC_READ | GENERIC_WRITE,
		0,
		NULL,
		CREATE_NEW,
		FILE_ATTRIBUTE_TEMPORARY,
		NULL);

	if (h == INVALID_HANDLE_VALUE) {
		DWORD err = GetLastError();
		printf("err %d\n", err);
		return 1;
	}

	printf("%p\n", h);

	return 0;
}

然后运行：

python .\child-gating1.py -f "C:\\Users\\l00379637\\source\\repos\\test_file_write\\Debug\\test_file_write.exe" -l log
✔ spawn(f=C:\\Users\\l00379637\\source\\repos\\test_file_write\\Debug\\test_file_write.exe)
✔ attach(pid=31252)
✔ enable_child_gating()
✔ create_script()
✔ load()
✔ resume(pid=31252)
err 80
⚡ message: pid=31252, payload={'type': 'CreateFileW', 'path': 'bar.txt'}
⚡ detached: pid=31252, reason='process-terminated'

可以看到

CreateFileW
OS API被hook获取了调用参数。

如果要hook其他API，修改JS部分代码即可：

Interceptor.attach(Module.findExportByName("kernel32.dll", "CreateFileW"), {
  onEnter: function (args) {
    //var fileNamePtr = args[0];
    //var fileName = Memory.readUtf16String(fileNamePtr);
    //send(fileName)
    send({
      type: 'CreateFileW',
      path: Memory.readUtf16String(args[0])
    });
  }
});