diff --git a/.gitignore b/.gitignore index 3179668..9f527d3 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,4 @@ *.pyc -.idea/ \ No newline at end of file +/*.egg-info/ +/build/ +/dist/ diff --git a/README.md b/README.md index 8af7ebf..0232ce2 100644 --- a/README.md +++ b/README.md @@ -7,10 +7,10 @@ Python bindings for [Frida](http://www.frida.re). To build and test your own egg, do something along the following lines: ``` -set FRIDA_VERSION=10.3.1.10.gd7c36fc # from c:\temp\frida\build\tmp-windows\frida-version.h -set FRIDA_EXTENSION=c:\temp\frida\build\frida-windows\Win32-Debug\lib\python2.7\site-packages\_frida.pyd -cd C:\Temp\frida\frida-python\src\ +set FRIDA_VERSION=12.0.0.10.gd7c36fc # from C:\src\frida\build\tmp-windows\frida-version.h +set FRIDA_EXTENSION=C:\src\frida\build\frida-windows\Win32-Debug\lib\python2.7\site-packages\_frida.pyd +cd C:\src\frida\frida-python\ python setup.py bdist_egg pip uninstall frida -easy_install dist\frida-10.3.1.10.gd7c36fc-py2.7-win32.egg +easy_install dist\frida-12.0.0.10.gd7c36fc-py2.7-win32.egg ``` diff --git a/examples/bytecode.py b/examples/bytecode.py index 84e5dd0..794e318 100644 --- a/examples/bytecode.py +++ b/examples/bytecode.py @@ -1,7 +1,9 @@ # -*- coding: utf-8 -*- from __future__ import print_function + import frida + system_session = frida.attach(0) bytecode = system_session.compile_script(name="bytecode-example", source="""\ 'use strict'; diff --git a/examples/child_gating.py b/examples/child_gating.py index f55ed5d..fabcebe 100644 --- a/examples/child_gating.py +++ b/examples/child_gating.py @@ -1,9 +1,12 @@ # -*- coding: utf-8 -*- from __future__ import print_function -import frida -from frida.application import Reactor + import threading +import frida +from frida_tools.application import Reactor + + class Application(object): def __init__(self): self._stop_requested = threading.Event() diff --git a/examples/detached.py b/examples/detached.py index 188b35e..dac552d 100644 --- a/examples/detached.py +++ b/examples/detached.py @@ -1,8 +1,11 @@ # -*- coding: utf-8 -*- from __future__ import print_function -import frida + import sys +import frida + + def on_detached(): print("on_detached") diff --git a/examples/inject_library/inject_blob.py b/examples/inject_library/inject_blob.py index 102df45..b567292 100644 --- a/examples/inject_library/inject_blob.py +++ b/examples/inject_library/inject_blob.py @@ -7,9 +7,12 @@ # from __future__ import unicode_literals, print_function -import frida + import sys +import frida + + def on_uninjected(id): print("on_uninjected id=%u" % id) diff --git a/examples/inject_library/inject_file.py b/examples/inject_library/inject_file.py index cf70afb..32fd008 100644 --- a/examples/inject_library/inject_file.py +++ b/examples/inject_library/inject_file.py @@ -7,9 +7,12 @@ # from __future__ import unicode_literals, print_function -import frida + import sys +import frida + + def on_uninjected(id): print("on_uninjected id=%u" % id) diff --git a/examples/rpc.py b/examples/rpc.py index 59dd57d..ffd4f88 100644 --- a/examples/rpc.py +++ b/examples/rpc.py @@ -1,7 +1,9 @@ # -*- coding: utf-8 -*- from __future__ import print_function + import frida + session = frida.attach("Twitter") script = session.create_script("""\ 'use strict'; diff --git a/frida-python-common.props b/frida-python-common.props index 4d766e3..d91d40e 100644 --- a/frida-python-common.props +++ b/frida-python-common.props @@ -17,34 +17,27 @@ mkdir $(OutDir)frida || exit 1 copy /B /Y "$(ProjectDir)src\frida\*.py" $(OutDir)frida || exit 1 echo. > $(IntDir)frida-package.stamp $(IntDir)frida-package.stamp - $(ProjectDir)src\frida\core.py;$(ProjectDir)src\frida\application.py;$(ProjectDir)src\frida\lsd.py;$(ProjectDir)src\frida\ps.py;$(ProjectDir)src\frida\kill.py;$(ProjectDir)src\frida\discoverer.py;$(ProjectDir)src\frida\tracer.py;$(ProjectDir)src\frida\repl.py + $(ProjectDir)src\frida\core.py rmdir /S /Q $(OutDir)frida 2>NUL mkdir $(OutDir)frida || exit 1 copy /B /Y "$(ProjectDir)src\frida\*.py" $(OutDir)frida || exit 1 echo. > $(IntDir)frida-package.stamp $(IntDir)frida-package.stamp - $(ProjectDir)src\frida\core.py;$(ProjectDir)src\frida\application.py;$(ProjectDir)src\frida\lsd.py;$(ProjectDir)src\frida\ps.py;$(ProjectDir)src\frida\kill.py;$(ProjectDir)src\frida\discoverer.py;$(ProjectDir)src\frida\tracer.py;$(ProjectDir)src\frida\repl.py + $(ProjectDir)src\frida\core.py rmdir /S /Q $(OutDir)frida 2>NUL mkdir $(OutDir)frida || exit 1 copy /B /Y "$(ProjectDir)src\frida\*.py" $(OutDir)frida || exit 1 echo. > $(IntDir)frida-package.stamp $(IntDir)frida-package.stamp - $(ProjectDir)src\frida\core.py;$(ProjectDir)src\frida\application.py;$(ProjectDir)src\frida\lsd.py;$(ProjectDir)src\frida\ps.py;$(ProjectDir)src\frida\kill.py;$(ProjectDir)src\frida\discoverer.py;$(ProjectDir)src\frida\tracer.py;$(ProjectDir)src\frida\repl.py + $(ProjectDir)src\frida\core.py rmdir /S /Q $(OutDir)frida 2>NUL mkdir $(OutDir)frida || exit 1 copy /B /Y "$(ProjectDir)src\frida\*.py" $(OutDir)frida || exit 1 echo. > $(IntDir)frida-package.stamp $(IntDir)frida-package.stamp - $(ProjectDir)src\frida\core.py;$(ProjectDir)src\frida\application.py;$(ProjectDir)src\frida\lsd.py;$(ProjectDir)src\frida\ps.py;$(ProjectDir)src\frida\kill.py;$(ProjectDir)src\frida\discoverer.py;$(ProjectDir)src\frida\tracer.py;$(ProjectDir)src\frida\repl.py + $(ProjectDir)src\frida\core.py - - - - - - - diff --git a/frida-python27.vcxproj.filters b/frida-python27.vcxproj.filters index d2e239a..d8a2ce8 100644 --- a/frida-python27.vcxproj.filters +++ b/frida-python27.vcxproj.filters @@ -26,27 +26,6 @@ Source Files\frida - - Source Files\frida - - - Source Files\frida - - - Source Files\frida - - - Source Files\frida - - - Source Files\frida - - - Source Files\frida - - - Source Files\frida - diff --git a/frida-python36.vcxproj.filters b/frida-python36.vcxproj.filters index 15f0456..842d22e 100644 --- a/frida-python36.vcxproj.filters +++ b/frida-python36.vcxproj.filters @@ -26,27 +26,6 @@ Source Files\frida - - Source Files\frida - - - Source Files\frida - - - Source Files\frida - - - Source Files\frida - - - Source Files\frida - - - Source Files\frida - - - Source Files\frida - diff --git a/src/frida/__init__.py b/frida/__init__.py similarity index 85% rename from src/frida/__init__.py rename to frida/__init__.py index ff4992e..12f138b 100644 --- a/src/frida/__init__.py +++ b/frida/__init__.py @@ -2,25 +2,23 @@ from __future__ import unicode_literals, print_function import threading + try: import _frida except Exception as ex: - import colorama - from colorama import Back, Fore, Style import sys - colorama.init(autoreset=True) print("") print("***") if str(ex).startswith("No module named "): - print(Back.RED + Fore.WHITE + Style.BRIGHT + "Frida native extension not found" + Style.RESET_ALL) - print(Fore.WHITE + Style.BRIGHT + "Please check your PYTHONPATH." + Style.RESET_ALL) + print("Frida native extension not found") + print("Please check your PYTHONPATH.") else: - print(Back.RED + Fore.WHITE + Style.BRIGHT + "Failed to load the Frida native extension: %s" % ex + Style.RESET_ALL) + print("Failed to load the Frida native extension: %s" % ex) if sys.version_info[0] == 2: current_python_version = "%d.%d" % sys.version_info[:2] else: current_python_version = "%d.x" % sys.version_info[0] - print(Fore.WHITE + Style.BRIGHT + "Please ensure that the extension was compiled for Python " + current_python_version + "." + Style.RESET_ALL) + print("Please ensure that the extension was compiled for Python " + current_python_version + ".") print("***") print("") raise ex @@ -44,39 +42,51 @@ NotSupportedError = _frida.NotSupportedError ProtocolError = _frida.ProtocolError TransportError = _frida.TransportError + def spawn(*args, **kwargs): return get_local_device().spawn(*args, **kwargs) + def resume(target): get_local_device().resume(target) + def kill(target): get_local_device().kill(target) + def attach(target): return get_local_device().attach(target) + def inject_library_file(target, path, entrypoint, data): return get_local_device().inject_library_file(target, path, entrypoint, data) + def inject_library_blob(target, blob, entrypoint, data): return get_local_device().inject_library_blob(target, blob, entrypoint, data) + def enumerate_devices(): return get_device_manager().enumerate_devices() + def get_local_device(): return _get_device(lambda device: device.type == 'local', timeout=0) + def get_remote_device(): return _get_device(lambda device: device.type == 'remote', timeout=0) + def get_usb_device(timeout = 0): return _get_device(lambda device: device.type == 'usb', timeout) + def get_device(id, timeout = 0): return _get_device(lambda device: device.id == id, timeout) + def _get_device(predicate, timeout): mgr = get_device_manager() def find_matching_device(): @@ -103,6 +113,7 @@ def _get_device(predicate, timeout): raise TimedOutError("timed out while waiting for device to appear") return device + def shutdown(): get_device_manager()._impl.close() diff --git a/src/frida/core.py b/frida/core.py similarity index 50% rename from src/frida/core.py rename to frida/core.py index d888349..fbc422f 100644 --- a/src/frida/core.py +++ b/frida/core.py @@ -1,8 +1,6 @@ # -*- coding: utf-8 -*- from __future__ import unicode_literals, print_function -import _frida -import bisect import fnmatch import json import numbers @@ -10,6 +8,8 @@ import sys import threading import traceback +import _frida + class DeviceManager(object): def __init__(self, impl): @@ -42,12 +42,14 @@ class DeviceManager(object): def off(self, signal, callback): self._impl.off(signal, callback) + class Device(object): def __init__(self, device): self.id = device.id self.name = device.name self.icon = device.icon self.type = device.type + self._impl = device def __repr__(self): @@ -125,30 +127,13 @@ class Device(object): else: return self.get_process(target).pid -class FunctionContainer(object): - def __init__(self): - self._functions = {} - """ - @param address is relative to container - """ - def ensure_function(self, address): - f = self._functions.get(address) - if f is not None: - return f - return self._do_ensure_function(address) - - def _do_ensure_function(self, address): - raise NotImplementedError("not implemented") - -class Session(FunctionContainer): +class Session(object): def __init__(self, impl): - super(Session, self).__init__() self._impl = impl - self._modules = None - self._module_map = None - self._script = None + def __repr__(self): + return repr(self._impl) def detach(self): self._impl.detach() @@ -159,42 +144,6 @@ class Session(FunctionContainer): def disable_child_gating(self): self._impl.disable_child_gating() - def enumerate_modules(self): - if self._modules is None: - raw_modules = self._get_api().enumerate_modules() - self._modules = [Module(data['name'], int(data['base'], 16), data['size'], data['path'], self) for data in raw_modules] - return self._modules - - def prefetch_modules(self): - modules = self.enumerate_modules() - pending = [m for m in modules if m._exports is None] - batches = self._get_api().enumerate_exports([m.path for m in pending]) - for i, raw_exports in enumerate(batches): - pending[i]._update_exports(raw_exports) - - """ - @param protection example '--x' - """ - def enumerate_ranges(self, protection): - raw_ranges = self._get_api().enumerate_ranges(protection) - return [Range(int(data['base'], 16), data['size'], data['protection']) for data in raw_ranges] - - def find_base_address(self, module_name): - raw_base_address = self._get_api().find_base_address(module_name) - return int(raw_base_address, 16) - - def read_bytes(self, address, size): - return self._get_api().read_byte_array("0x%x" % address, size) - - def write_bytes(self, address, data): - self._get_api().write_byte_array("0x%x" % address, [x for x in iterbytes(data)]) - - def read_utf8(self, address, length=-1): - return self._get_api().read_utf8("0x%x" % address, length) - - def write_utf8(self, address, string): - self._get_api().write_utf8("0x%x" % address, string) - def create_script(self, *args, **kwargs): return Script(self._impl.create_script(*args, **kwargs)) @@ -219,87 +168,13 @@ class Session(FunctionContainer): def off(self, signal, callback): self._impl.off(signal, callback) - def _get_api(self): - return self._get_script().exports - - def _get_script(self): - if self._script is None: - self._script = self.create_script(name="session", source=self._create_session_script()) - self._script.on('message', self._on_message) - self._script.load() - return self._script - - def _on_message(self, message, data): - print("[session]", message, data) - - def _create_session_script(self): - return """\ -"use strict"; - -rpc.exports = { - enumerateModules: function () { - return Process.enumerateModulesSync(); - }, - enumerateExports: function (modulePaths) { - return modulePaths.map(function (modulePath) { - return Module.enumerateExportsSync(modulePath); - }); - }, - enumerateRanges: function (protection) { - return Process.enumerateRangesSync(protection); - }, - findBaseAddress: function (moduleName) { - var address = Module.findBaseAddress(moduleName); - return (address !== null) ? address.toString() : "0"; - }, - readByteArray: function (address, size) { - return Memory.readByteArray(ptr(address), size); - }, - writeByteArray: function (address, data) { - var base = ptr(address); - for (var i = 0; i !== data.length; i++) { - Memory.writeU8(base.add(i), data[i]); - } - }, - readUtf8: function (address, length) { - return Memory.readUtf8String(ptr(address), length); - }, - writeUtf8: function (address, string) { - Memory.writeUtf8String(ptr(address), string); - }, - enumerateModuleExports: function (modulePath) { - return Module.enumerateExportsSync(modulePath).filter(function (e) { - return e.type === 'function'; - }); - }, - enumerateModuleRanges: function (modulePath, protection) { - return Module.enumerateRangesSync(modulePath, protection); - } -}; -""" - - def _do_ensure_function(self, absolute_address): - if self._module_map is None: - self._module_map = ModuleMap(self.enumerate_modules()) - m = self._module_map.lookup(absolute_address) - if m is not None: - f = m.ensure_function(absolute_address - m.base_address) - else: - f = Function("dsub_%x" % absolute_address, absolute_address) - self._functions[absolute_address] = f - return f - - def __getattr__(self, attr): - if attr == 'session': - raise KeyError("Please update your code from `.session.create_script()` to `.create_script()`") - else: - return getattr(super(Session, self), attr) class Script(object): def __init__(self, impl): self.exports = ScriptExports(self) self._impl = impl + self._on_message_callbacks = [] self._log_handler = self._on_log @@ -427,10 +302,12 @@ class Script(object): else: print(text, file=sys.stderr) + class RPCException(Exception): def __str__(self): return self.args[2] if len(self.args) >= 3 else self.args[0] + class ScriptExports(object): def __init__(self, script): self._script = script @@ -442,142 +319,10 @@ class ScriptExports(object): return script._rpc_request('call', js_name, args) return method -class Module(FunctionContainer): - def __init__(self, name, base_address, size, path, session): - super(Module, self).__init__() - self.name = name - self.base_address = base_address - self.size = size - self.path = path - self._exports = None - self._session = session - - def __repr__(self): - return "Module(name=\"%s\", base_address=0x%x, size=%d, path=\"%s\")" % (self.name, self.base_address, self.size, self.path) - - def __hash__(self): - return self.base_address.__hash__() - - def __cmp__(self, other): - return self.base_address.__cmp__(other.base_address) - - def __eq__(self, other): - return self.base_address == other.base_address - - def __ne__(self, other): - return self.base_address != other.base_address - - def enumerate_exports(self): - if self._exports is None: - raw_exports = self._session._get_api().enumerate_module_exports(self.path) - self._update_exports(raw_exports) - return self._exports - - """ - @param protection example '--x' - """ - def enumerate_ranges(self, protection): - raw_ranges = self._session._get_script().exports.enumerate_module_ranges(self.path, protection) - return [Range(int(data['base'], 16), data['size'], data['protection']) for data in raw_ranges] - - def _update_exports(self, exports): - self._exports = [] - for export in exports: - relative_address = int(export["address"], 16) - self.base_address - mf = ModuleFunction(self, export["name"], relative_address, True) - self._exports.append(mf) - self._functions[relative_address] = mf - - def _do_ensure_function(self, relative_address): - self.enumerate_exports() - mf = self._functions.get(relative_address) - if mf is None: - mf = ModuleFunction(self, "sub_%x" % relative_address, relative_address, False) - self._functions[relative_address] = mf - return mf - -class Function(object): - def __init__(self, name, absolute_address): - self.name = name - self.absolute_address = absolute_address - - def __str__(self): - return self.name - - def __repr__(self): - return "Function(name=\"%s\", absolute_address=0x%x)" % (self.name, self.absolute_address) - - def __hash__(self): - return self.absolute_address.__hash__() - - def __cmp__(self, other): - return self.absolute_address.__cmp__(other.absolute_address) - - def __eq__(self, other): - return self.absolute_address == other.absolute_address - - def __ne__(self, other): - return self.absolute_address != other.absolute_address - -class ModuleFunction(Function): - def __init__(self, module, name, relative_address, exported): - super(ModuleFunction, self).__init__(name, module.base_address + relative_address) - self.module = module - self.relative_address = relative_address - self.exported = exported - - def __repr__(self): - return "ModuleFunction(module=\"%s\", name=\"%s\", relative_address=0x%x)" % (self.module.name, self.name, self.relative_address) - -class ObjCMethod(Function): - def __init__(self, mtype, cls, method, address): - self.mtype = mtype - self.cls = cls - self.method = method - self.address = address - super(ObjCMethod, self).__init__(self.display_name(), address) - - def display_name(self): - return '{mtype}[{cls} {method}]'.format(mtype=self.mtype, cls=self.cls, method=self.method) - - def __repr__(self): - return "ObjCMethod(mtype=\"%s\", cls=\"%s\", method=\"%s\", address=0x%x)" % (self.mtype, self.cls, self.method, self.address) - -class Range(object): - def __init__(self, base_address, size, protection): - self.base_address = base_address - self.size = size - self.protection = protection - - def __repr__(self): - return "Range(base_address=0x%x, size=%s, protection='%s')" % (self.base_address, self.size, self.protection) class Error(Exception): pass -class AddressMap(object): - def __init__(self, items, get_address, get_size): - self._items = sorted(items, key=get_address) - self._indices = [ get_address(item) for item in self._items ] - self._get_address = get_address - self._get_size = get_size - - def lookup(self, address): - index = bisect.bisect(self._indices, address) - if index == 0: - return None - item = self._items[index - 1] - if address >= self._get_address(item) + self._get_size(item): - return None - return item - -class ModuleMap(AddressMap): - def __init__(self, modules): - super(ModuleMap, self).__init__(modules, lambda m: m.base_address, lambda m: m.size) - -class FunctionMap(AddressMap): - def __init__(self, functions, get_address=lambda f: f.absolute_address): - super(FunctionMap, self).__init__(functions, get_address, lambda f: 1) def _to_camel_case(name): result = "" @@ -592,10 +337,8 @@ def _to_camel_case(name): result += c.lower() return result + if sys.version_info[0] >= 3: string_types = str, - iterbytes = lambda x: iter(x) else: string_types = basestring, - def iterbytes(data): - return (ord(char) for char in data) diff --git a/src/frida/meson.build b/frida/meson.build similarity index 56% rename from src/frida/meson.build rename to frida/meson.build index a5c061f..0b36e11 100644 --- a/src/frida/meson.build +++ b/frida/meson.build @@ -1,13 +1,6 @@ frida_sources = [ '__init__.py', - 'application.py', 'core.py', - 'discoverer.py', - 'kill.py', - 'lsd.py', - 'ps.py', - 'repl.py', - 'tracer.py', ] install_data(frida_sources, install_dir: join_paths(python_site_packages, 'frida')) diff --git a/meson.build b/meson.build index e0fac22..35a6c97 100644 --- a/meson.build +++ b/meson.build @@ -75,3 +75,4 @@ configure_file(input: 'config.h.in', configuration: cdata) subdir('src') +subdir('frida') diff --git a/src/setup.py b/setup.py similarity index 89% rename from src/setup.py rename to setup.py index 6c9f943..1d388bc 100755 --- a/src/setup.py +++ b/setup.py @@ -1,6 +1,6 @@ # -*- coding: utf-8 -*- - from __future__ import print_function + try: from io import BytesIO except: @@ -8,11 +8,11 @@ except: from cStringIO import StringIO as BytesIO except: from StringIO import StringIO as BytesIO +import os +import platform from setuptools import setup from setuptools.command.build_ext import build_ext from setuptools.extension import Extension -import os -import platform import shutil import struct import sys @@ -20,11 +20,12 @@ try: from urllib.request import urlopen, Request except: from urllib2 import urlopen, Request -import zipfile try: import xmlrpclib except ImportError: import xmlrpc.client as xmlrpclib +import zipfile + package_dir = os.path.dirname(os.path.realpath(__file__)) pkg_info = os.path.join(package_dir, "PKG-INFO") @@ -39,6 +40,8 @@ else: frida_version = os.environ['FRIDA_VERSION'] long_description = open(os.path.join(root_dir, "README.md")).read() frida_extension = os.environ['FRIDA_EXTENSION'] +frida_major_version = int(frida_version.split(".")[0]) + class UrllibTransport(xmlrpclib.Transport): def __init__(self, *args, **kwargs): @@ -52,6 +55,7 @@ class UrllibTransport(xmlrpclib.Transport): fp = urlopen(req) return self.parse_response(fp) + class FridaPrebuiltExt(build_ext): def build_extension(self, ext): target = self.get_ext_fullpath(ext.name) @@ -132,23 +136,19 @@ class FridaPrebuiltExt(build_ext): else: shutil.copyfile(frida_extension, target) + setup( - name='frida', + name="frida", version=frida_version, - description="Inject JavaScript to explore native apps on Windows, macOS, Linux, iOS, Android, and QNX", + description="Dynamic instrumentation toolkit for developers, reverse-engineers, and security researchers", long_description=long_description, long_description_content_type="text/markdown", author="Frida Developers", - author_email="oleavr@nowsecure.com", + author_email="oleavr@frida.re", url="https://www.frida.re", - install_requires=[ - "colorama >= 0.2.7, < 1.0.0", - "prompt-toolkit >= 0.57, < 2.0.0", - "pygments >= 2.0.2, < 3.0.0" - ], license="wxWindows Library Licence, Version 3.1", zip_safe=True, - keywords="frida debugger inject javascript windows macos linux ios iphone ipad android", + keywords="frida debugger dynamic instrumentation inject javascript windows macos linux ios iphone ipad android qnx", classifiers=[ "Development Status :: 5 - Production/Stable", "Environment :: Console", @@ -173,16 +173,6 @@ setup( "Topic :: Software Development :: Libraries :: Python Modules" ], packages=['frida'], - entry_points={ - 'console_scripts': [ - "frida = frida.repl:main", - "frida-discover = frida.discoverer:main", - "frida-ls-devices = frida.lsd:main", - "frida-ps = frida.ps:main", - "frida-kill = frida.kill:main", - "frida-trace = frida.tracer:main" - ] - }, ext_modules=[Extension('_frida', [])], cmdclass={ 'build_ext': FridaPrebuiltExt diff --git a/src/frida/application.py b/src/frida/application.py deleted file mode 100644 index 6fe5334..0000000 --- a/src/frida/application.py +++ /dev/null @@ -1,437 +0,0 @@ -# -*- coding: utf-8 -*- -from __future__ import unicode_literals, print_function - -import collections -import errno -from optparse import OptionParser -import os -import platform -import select -import signal -import sys -import threading -from time import time - -import colorama -from colorama import Fore, Style - -import frida - -if platform.system() == 'Windows': - import msvcrt - -def input_with_timeout(timeout): - if platform.system() == 'Windows': - start_time = time() - s = '' - while True: - while msvcrt.kbhit(): - c = msvcrt.getche() - if ord(c) == 13: # enter_key - break - elif ord(c) >= 32: #space_char - s += c.decode('utf-8') - if time() - start_time > timeout: - return None - - return s - else: - while True: - try: - rlist, _, _ = select.select([sys.stdin], [], [], timeout) - break - except (OSError, select.error) as e: - if e.args[0] != errno.EINTR: - raise e - if rlist: - return sys.stdin.readline() - else: - return None - -def await_enter(reactor): - try: - while input_with_timeout(0.5) == None: - if not reactor.is_running(): - break - except KeyboardInterrupt: - print('') - -class ConsoleState: - EMPTY = 1 - STATUS = 2 - TEXT = 3 - -class ConsoleApplication(object): - def __init__(self, run_until_return=await_enter, on_stop=None): - colorama.init() - - parser = OptionParser(usage=self._usage(), version=frida.__version__) - - if self._needs_device(): - parser.add_option("-D", "--device", help="connect to device with the given ID", - metavar="ID", type='string', action='store', dest="device_id", default=None) - parser.add_option("-U", "--usb", help="connect to USB device", - action='store_const', const='usb', dest="device_type", default=None) - parser.add_option("-R", "--remote", help="connect to remote frida-server", - action='store_const', const='remote', dest="device_type", default=None) - parser.add_option("-H", "--host", help="connect to remote frida-server on HOST", - metavar="HOST", type='string', action='store', dest="host", default=None) - - if self._needs_target(): - def store_target(option, opt_str, target_value, parser, target_type, *args, **kwargs): - if target_type == 'file': - target_value = [target_value] - setattr(parser.values, 'target', (target_type, target_value)) - parser.add_option("-f", "--file", help="spawn FILE", metavar="FILE", - type='string', action='callback', callback=store_target, callback_args=('file',)) - parser.add_option("-n", "--attach-name", help="attach to NAME", metavar="NAME", - type='string', action='callback', callback=store_target, callback_args=('name',)) - parser.add_option("-p", "--attach-pid", help="attach to PID", metavar="PID", - type='int', action='callback', callback=store_target, callback_args=('pid',)) - parser.add_option("--debug", help="enable the Node.js compatible script debugger", - action='store_true', dest="enable_debugger", default=False) - parser.add_option("--enable-jit", help="enable JIT", - action='store_true', dest="enable_jit", default=False) - - self._add_options(parser) - - (options, args) = parser.parse_args() - - if sys.version_info[0] < 3: - input_encoding = sys.stdin.encoding or 'UTF-8' - args = [arg.decode(input_encoding) for arg in args] - - if self._needs_device(): - self._device_id = options.device_id - self._device_type = options.device_type - self._host = options.host - else: - self._device_id = None - self._device_type = None - self._host = None - self._device = None - self._schedule_on_output = lambda pid, fd, data: self._reactor.schedule(lambda: self._on_output(pid, fd, data)) - self._schedule_on_device_lost = lambda: self._reactor.schedule(self._on_device_lost) - self._spawned_pid = None - self._spawned_argv = None - self._session = None - if self._needs_target(): - self._enable_debugger = options.enable_debugger - self._enable_jit = options.enable_jit - else: - self._enable_debugger = False - self._enable_jit = False - self._schedule_on_session_detached = lambda reason: self._reactor.schedule(lambda: self._on_session_detached(reason)) - self._started = False - self._resumed = False - self._reactor = Reactor(run_until_return, on_stop) - self._exit_status = None - self._console_state = ConsoleState.EMPTY - self._have_terminal = sys.stdin.isatty() and sys.stdout.isatty() and not os.environ.get("TERM", '') == "dumb" - self._quiet = False - if sum(map(lambda v: int(v is not None), (self._device_id, self._device_type, self._host))) > 1: - parser.error("Only one of -D, -U, -R, and -H may be specified") - - if self._needs_target(): - target = getattr(options, 'target', None) - if target is None: - if len(args) < 1: - parser.error("target file, process name or pid must be specified") - target = infer_target(args[0]) - args.pop(0) - target = expand_target(target) - if target[0] == 'file': - argv = target[1] - argv.extend(args) - args = [] - self._target = target - else: - self._target = None - - self._initialize(parser, options, args) - - def run(self): - mgr = frida.get_device_manager() - - on_devices_changed = lambda: self._reactor.schedule(self._try_start) - mgr.on('changed', on_devices_changed) - - self._reactor.schedule(self._try_start) - self._reactor.schedule(self._show_message_if_no_device, delay=1) - - old_sigterm_handler = signal.signal(signal.SIGTERM, lambda n, f: self._exit(0)) - - self._reactor.run() - - signal.signal(signal.SIGTERM, old_sigterm_handler) - - if self._started: - self._stop() - - if self._session is not None: - self._session.off('detached', self._schedule_on_session_detached) - self._session.detach() - self._session = None - - if self._spawned_pid is not None: - try: - self._device.kill(self._spawned_pid) - except: - pass - - if self._device is not None: - self._device.off('output', self._schedule_on_output) - self._device.off('lost', self._schedule_on_device_lost) - - mgr.off('changed', on_devices_changed) - - frida.shutdown() - sys.exit(self._exit_status) - - def _add_options(self, parser): - pass - - def _initialize(self, parser, options, args): - pass - - def _needs_device(self): - return True - - def _needs_target(self): - return False - - def _start(self): - pass - - def _stop(self): - pass - - def _resume(self): - if self._resumed: - return - if self._spawned_pid is not None: - self._device.resume(self._spawned_pid) - self._resumed = True - - def _exit(self, exit_status): - self._exit_status = exit_status - self._reactor.stop() - - def _try_start(self): - if self._device is not None: - return - if self._device_id is not None: - try: - self._device = frida.get_device(self._device_id) - except: - self._update_status("Device '%s' not found" % self._device_id) - self._exit(1) - return - elif self._device_type is not None: - self._device = find_device(self._device_type) - if self._device is None: - return - elif self._host is not None: - self._device = frida.get_device_manager().add_remote_device(self._host) - else: - self._device = frida.get_local_device() - self._device.on('output', self._schedule_on_output) - self._device.on('lost', self._schedule_on_device_lost) - if self._target is not None: - spawning = True - try: - target_type, target_value = self._target - if target_type == 'file': - argv = target_value - if not self._quiet: - self._update_status("Spawning `%s`..." % " ".join(argv)) - self._spawned_pid = self._device.spawn(argv) - self._spawned_argv = argv - attach_target = self._spawned_pid - else: - attach_target = target_value - if not self._quiet: - self._update_status("Attaching...") - spawning = False - self._session = self._device.attach(attach_target) - if self._enable_jit: - self._session.enable_jit() - if self._enable_debugger: - self._session.enable_debugger() - self._print("Debugger listening on port 5858\n") - self._session.on('detached', self._schedule_on_session_detached) - except Exception as e: - if spawning: - self._update_status("Failed to spawn: %s" % e) - else: - self._update_status("Failed to attach: %s" % e) - self._exit(1) - return - self._start() - self._started = True - - def _show_message_if_no_device(self): - if self._device is None: - self._print("Waiting for USB device to appear...") - - def _on_output(self, pid, fd, data): - if data is None: - return - if fd == 1: - prefix = "stdout> " - stream = sys.stdout - else: - prefix = "stderr> " - stream = sys.stderr - encoding = stream.encoding or 'UTF-8' - text = data.decode(encoding, errors='replace') - if text.endswith("\n"): - text = text[:-1] - lines = text.split("\n") - self._print(prefix + ("\n" + prefix).join(lines)) - - def _on_device_lost(self): - if self._exit_status is not None: - return - self._print("Device disconnected.") - self._exit(1) - - def _on_session_detached(self, reason): - message = reason[0].upper() + reason[1:].replace("-", " ") - self._print(Fore.RED + Style.BRIGHT + message + Style.RESET_ALL) - self._exit(1) - - def _clear_status(self): - if self._console_state == ConsoleState.STATUS: - print("\033[A" + (80 * " ")) - - def _update_status(self, message): - if self._have_terminal: - if self._console_state == ConsoleState.STATUS: - cursor_position = "\033[A" - else: - cursor_position = "" - print("%-80s" % (cursor_position + Style.BRIGHT + message + Style.RESET_ALL,)) - self._console_state = ConsoleState.STATUS - else: - print(Style.BRIGHT + message + Style.RESET_ALL) - - def _print(self, *args, **kwargs): - encoded_args = [] - if sys.version_info[0] >= 3: - string_type = str - decoder = "unicode-escape" - else: - string_type = unicode - decoder = "string-escape" - encoding = sys.stdout.encoding or 'UTF-8' - for arg in args: - if isinstance(arg, string_type): - encoded_args.append(arg.encode(encoding, errors='replace').decode(decoder, errors='replace')) - else: - encoded_args.append(arg) - print(*encoded_args, **kwargs) - self._console_state = ConsoleState.TEXT - - def _log(self, level, text): - if level == 'info': - self._print(text) - else: - color = Fore.RED if level == 'error' else Fore.YELLOW - self._print(color + Style.BRIGHT + text + Style.RESET_ALL) - -def find_device(type): - for device in frida.enumerate_devices(): - if device.type == type: - return device - return None - -def infer_target(target_value): - if target_value.startswith('.') or target_value.startswith(os.path.sep) \ - or (platform.system() == 'Windows' \ - and target_value[0].isalpha() \ - and target_value[1] == ":" \ - and target_value[2] == "\\"): - target_type = 'file' - target_value = [target_value] - else: - try: - target_value = int(target_value) - target_type = 'pid' - except: - target_type = 'name' - return (target_type, target_value) - -def expand_target(target): - target_type, target_value = target - if target_type == 'file': - target_value = [target_value[0]] - return (target_type, target_value) - - -class Reactor(object): - def __init__(self, run_until_return, on_stop=None): - self._running = False - self._run_until_return = run_until_return - self._on_stop = on_stop - self._pending = collections.deque([]) - self._lock = threading.Lock() - self._cond = threading.Condition(self._lock) - - def is_running(self): - with self._lock: - return self._running - - def run(self): - with self._lock: - self._running = True - - worker = threading.Thread(target=self._run) - worker.start() - - self._run_until_return(self) - - self.stop() - worker.join() - - def _run(self): - running = True - while running: - now = time() - work = None - timeout = None - previous_pending_length = -1 - with self._lock: - for item in self._pending: - (f, when) = item - if now >= when: - work = f - self._pending.remove(item) - break - if len(self._pending) > 0: - timeout = max([min(map(lambda item: item[1], self._pending)) - now, 0]) - previous_pending_length = len(self._pending) - if work is not None: - work() - with self._lock: - if self._running and len(self._pending) == previous_pending_length: - self._cond.wait(timeout) - running = self._running - - if self._on_stop is not None: - self._on_stop() - - def stop(self): - with self._lock: - self._running = False - self._cond.notify() - - def schedule(self, f, delay=None): - now = time() - if delay is not None: - when = now + delay - else: - when = now - with self._lock: - self._pending.append((f, when)) - self._cond.notify() diff --git a/src/frida/discoverer.py b/src/frida/discoverer.py deleted file mode 100644 index 4bee0db..0000000 --- a/src/frida/discoverer.py +++ /dev/null @@ -1,203 +0,0 @@ -# -*- coding: utf-8 -*- -from __future__ import unicode_literals, print_function - -from frida.application import await_enter -from frida.core import ModuleFunction -import threading - - -class Discoverer(object): - def __init__(self, reactor): - self._reactor = reactor - self._script = None - - def dispose(self): - if self._script is not None: - try: - self._script.unload() - except: - pass - self._script = None - - def start(self, session, ui): - def on_message(message, data): - self._reactor.schedule(lambda: self._process_message(message, data, session, ui)) - session.enable_jit() - self._script = session.create_script(name="discoverer", source=self._create_discover_script()) - self._script.on('message', on_message) - self._script.load() - - def stop(self): - self._script.post({ - 'to': "/sampler", - 'name': '+stop', - 'payload': {} - }) - - def _create_discover_script(self): - return """"use strict";\ - -function Sampler() { - var threadIds = []; - var result = {}; - - function onStanza(stanza) { - if (stanza.to === "/sampler") { - if (stanza.name === '+stop') - stop(); - } - - recv(onStanza); - } - - this.start = function () { - threadIds = []; - Process.enumerateThreads({ - onMatch: function (thread) { - threadIds.push(thread.id); - }, - onComplete: function () { - threadIds.forEach(function (threadId) { - Stalker.follow(threadId, { - events: { call: true }, - onCallSummary: function (summary) { - for (var address in summary) { - if (summary.hasOwnProperty(address)) { - var count = result[address] || 0; - result[address] = count + summary[address]; - } - } - } - }); - }); - - send({ - from: "/sampler", - name: '+started', - payload: { - total: threadIds.length - } - }); - } - }); - }; - - function stop() { - threadIds.forEach(function (threadId) { - Stalker.unfollow(threadId); - }); - threadIds = []; - - send({ - from: "/sampler", - name: '+stopped', - payload: { - result: result - } - }); - result = {}; - } - - recv(onStanza); -}; - -var sampler = new Sampler(); -setTimeout(function () { sampler.start(); }, 0); -""" - - def _process_message(self, message, data, session, ui): - if message['type'] == 'send': - stanza = message['payload'] - name = stanza['name'] - payload = stanza['payload'] - if stanza['from'] == "/sampler": - if name == '+started': - ui.on_sample_start(payload['total']) - elif name == '+stopped': - module_functions = {} - dynamic_functions = [] - for address, count in payload['result'].items(): - address = int(address, 16) - function = session.ensure_function(address) - if isinstance(function, ModuleFunction): - functions = module_functions.get(function.module, []) - if len(functions) == 0: - module_functions[function.module] = functions - functions.append((function, count)) - else: - dynamic_functions.append((function, count)) - ui.on_sample_result(module_functions, dynamic_functions) - else: - print(message, data) - else: - print(message, data) - else: - print(message, data) - -class UI(object): - def on_sample_start(self, total): - pass - - def on_sample_result(self, module_functions, dynamic_functions): - pass - - -def main(): - from frida.application import ConsoleApplication - - class DiscovererApplication(ConsoleApplication, UI): - def __init__(self): - self._results_received = threading.Event() - ConsoleApplication.__init__(self, self._await_keys) - - def _await_keys(self, reactor): - await_enter(reactor) - reactor.schedule(lambda: self._discoverer.stop()) - while reactor.is_running() and not self._results_received.is_set(): - self._results_received.wait(0.5) - - def _usage(self): - return "usage: %prog [options] target" - - def _initialize(self, parser, options, args): - self._discoverer = None - - def _needs_target(self): - return True - - def _start(self): - self._update_status("Injecting script...") - self._discoverer = Discoverer(self._reactor) - self._discoverer.start(self._session, self) - - def _stop(self): - self._print("Stopping...") - self._discoverer.dispose() - self._discoverer = None - - def on_sample_start(self, total): - self._update_status("Tracing %d threads. Press ENTER to stop." % total) - self._resume() - - def on_sample_result(self, module_functions, dynamic_functions): - for module, functions in module_functions.items(): - self._print(module.name) - self._print("\t%-10s\t%s" % ("Calls", "Function")) - for function, count in sorted(functions, key=lambda item: item[1], reverse=True): - self._print("\t%-10d\t%s" % (count, function)) - self._print("") - - if len(dynamic_functions) > 0: - self._print("Dynamic functions:") - self._print("\t%-10s\t%s" % ("Calls", "Function")) - for function, count in sorted(dynamic_functions, key=lambda item: item[1], reverse=True): - self._print("\t%-10d\t%s" % (count, function)) - - self._results_received.set() - - app = DiscovererApplication() - app.run() - - -if __name__ == '__main__': - main() diff --git a/src/frida/kill.py b/src/frida/kill.py deleted file mode 100644 index 5f125e1..0000000 --- a/src/frida/kill.py +++ /dev/null @@ -1,35 +0,0 @@ -# -*- coding: utf-8 -*- -from __future__ import print_function - - -def main(): - import frida - from frida.application import ConsoleApplication, infer_target, expand_target - - class KillApplication(ConsoleApplication): - def _usage(self): - return "usage: %prog [options] process" - - def _start(self): - try: - self._device.kill(self._process) - except frida.ProcessNotFoundError: - self._update_status('unable to find process: %s' % self._process) - self._exit(1) - self._exit(0) - - def _initialize(self, parser, options, args): - if len(args) < 1: - parser.error('process name or pid must be specified') - process = expand_target(infer_target(args[0])) - if process[0] == 'file': - parser.error('process name or pid must be specified') - - self._process = process[1] - - app = KillApplication() - app.run() - - -if __name__ == '__main__': - main() diff --git a/src/frida/lsd.py b/src/frida/lsd.py deleted file mode 100644 index c838385..0000000 --- a/src/frida/lsd.py +++ /dev/null @@ -1,88 +0,0 @@ -# -*- coding: utf-8 -*- -from __future__ import unicode_literals, print_function - -def main(): - import frida - from frida.application import ConsoleApplication - - class LSDApplication(ConsoleApplication): - def _usage(self): - return "usage: %prog [options]" - - def _needs_device(self): - return False - - def _start(self): - try: - devices = frida.enumerate_devices() - except Exception as e: - self._update_status("Failed to enumerate devices: %s" % e) - self._exit(1) - return - id_column_width = max(map(lambda device: len(device.id), devices)) - type_column_width = max(map(lambda device: len(device.type), devices)) - name_column_width = max(map(lambda device: len(device.name), devices)) - header_format = "%-" + str(id_column_width) + "s " + \ - "%-" + str(type_column_width) + "s " + \ - "%-" + str(name_column_width) + "s" - self._print(header_format % ("Id", "Type", "Name")) - self._print("%s %s %s" % (id_column_width * "-", type_column_width * "-", name_column_width * "-")) - line_format = "%-" + str(id_column_width) + "s " + \ - "%-" + str(type_column_width) + "s " + \ - "%-" + str(name_column_width) + "s" - for device in sorted(devices, key=cmp_to_key(compare_devices)): - self._print(line_format % (device.id, device.type, device.name)) - self._exit(0) - - def compare_devices(a, b): - a_score = score(a) - b_score = score(b) - if a_score == b_score: - if a.name > b.name: - return 1 - elif a.name < b.name: - return -1 - else: - return 0 - else: - if a_score > b_score: - return -1 - elif a_score < b_score: - return 1 - else: - return 0 - - def score(device): - type = device.type - if type == 'local': - return 3 - elif type == 'usb': - return 2 - else: - return 1 - - def cmp_to_key(mycmp): - "Convert a cmp= function into a key= function" - class K: - def __init__(self, obj, *args): - self.obj = obj - def __lt__(self, other): - return mycmp(self.obj, other.obj) < 0 - def __gt__(self, other): - return mycmp(self.obj, other.obj) > 0 - def __eq__(self, other): - return mycmp(self.obj, other.obj) == 0 - def __le__(self, other): - return mycmp(self.obj, other.obj) <= 0 - def __ge__(self, other): - return mycmp(self.obj, other.obj) >= 0 - def __ne__(self, other): - return mycmp(self.obj, other.obj) != 0 - return K - - device = LSDApplication() - device.run() - - -if __name__ == '__main__': - main() diff --git a/src/frida/ps.py b/src/frida/ps.py deleted file mode 100644 index 2ef2302..0000000 --- a/src/frida/ps.py +++ /dev/null @@ -1,125 +0,0 @@ -# -*- coding: utf-8 -*- -from __future__ import unicode_literals, print_function - -def main(): - from frida.application import ConsoleApplication - - class PSApplication(ConsoleApplication): - def _add_options(self, parser): - parser.add_option("-a", "--applications", help="list only applications", - action='store_true', dest="list_only_applications", default=False) - parser.add_option("-i", "--installed", help="include all installed applications", - action='store_true', dest="include_all_applications", default=False) - - def _initialize(self, parser, options, args): - if options.include_all_applications and not options.list_only_applications: - parser.error("-i cannot be used without -a") - self._list_only_applications = options.list_only_applications - self._include_all_applications = options.include_all_applications - - def _usage(self): - return "usage: %prog [options]" - - def _start(self): - if self._list_only_applications: - try: - applications = self._device.enumerate_applications() - except Exception as e: - self._update_status("Failed to enumerate applications: %s" % e) - self._exit(1) - return - if not self._include_all_applications: - applications = list(filter(lambda app: app.pid != 0, applications)) - if len(applications) > 0: - pid_column_width = max(map(lambda app: len("%d" % app.pid), applications)) - name_column_width = max(map(lambda app: len(app.name), applications)) - identifier_column_width = max(map(lambda app: len(app.identifier), applications)) - else: - pid_column_width = 0 - name_column_width = 0 - identifier_column_width = 0 - header_format = "%" + str(pid_column_width) + "s " + \ - "%-" + str(name_column_width) + "s " + \ - "%-" + str(identifier_column_width) + "s" - self._print(header_format % ("PID", "Name", "Identifier")) - self._print("%s %s %s" % (pid_column_width * "-", name_column_width * "-", identifier_column_width * "-")) - line_format = "%" + str(pid_column_width) + "s " + \ - "%-" + str(name_column_width) + "s " + \ - "%-" + str(identifier_column_width) + "s" - for app in sorted(applications, key=cmp_to_key(compare_applications)): - if app.pid == 0: - self._print(line_format % ("-", app.name, app.identifier)) - else: - self._print(line_format % (app.pid, app.name, app.identifier)) - else: - try: - processes = self._device.enumerate_processes() - except Exception as e: - self._update_status("Failed to enumerate processes: %s" % e) - self._exit(1) - return - pid_column_width = max(map(lambda p: len("%d" % p.pid), processes)) - name_column_width = max(map(lambda p: len(p.name), processes)) - header_format = "%" + str(pid_column_width) + "s %s" - self._print(header_format % ("PID", "Name")) - self._print("%s %s" % (pid_column_width * "-", name_column_width * "-")) - line_format = "%" + str(pid_column_width) + "d %s" - for process in sorted(processes, key=cmp_to_key(compare_processes)): - self._print(line_format % (process.pid, process.name)) - self._exit(0) - - def compare_applications(a, b): - a_is_running = a.pid != 0 - b_is_running = b.pid != 0 - if a_is_running == b_is_running: - if a.name > b.name: - return 1 - elif a.name < b.name: - return -1 - else: - return 0 - elif a_is_running: - return -1 - else: - return 1 - - def compare_processes(a, b): - a_has_icon = a.get_small_icon() is not None - b_has_icon = b.get_small_icon() is not None - if a_has_icon == b_has_icon: - if a.name > b.name: - return 1 - elif a.name < b.name: - return -1 - else: - return 0 - elif a_has_icon: - return -1 - else: - return 1 - - def cmp_to_key(mycmp): - "Convert a cmp= function into a key= function" - class K: - def __init__(self, obj, *args): - self.obj = obj - def __lt__(self, other): - return mycmp(self.obj, other.obj) < 0 - def __gt__(self, other): - return mycmp(self.obj, other.obj) > 0 - def __eq__(self, other): - return mycmp(self.obj, other.obj) == 0 - def __le__(self, other): - return mycmp(self.obj, other.obj) <= 0 - def __ge__(self, other): - return mycmp(self.obj, other.obj) >= 0 - def __ne__(self, other): - return mycmp(self.obj, other.obj) != 0 - return K - - app = PSApplication() - app.run() - - -if __name__ == '__main__': - main() diff --git a/src/frida/repl.py b/src/frida/repl.py deleted file mode 100755 index 4c7b657..0000000 --- a/src/frida/repl.py +++ /dev/null @@ -1,765 +0,0 @@ -# -*- coding: utf-8 -*- -from __future__ import unicode_literals, print_function -import signal -import threading - -def main(): - import codecs - from colorama import Fore, Style - import frida - from frida.application import ConsoleApplication - import hashlib - import json - import os - import platform - from prompt_toolkit.shortcuts import create_prompt_application, create_output, create_eventloop - from prompt_toolkit.history import FileHistory - from prompt_toolkit.completion import Completion, Completer - from prompt_toolkit.interface import CommandLineInterface - from pygments.lexers import JavascriptLexer - from pygments.token import Token - import re - import sys - try: - from urllib.request import build_opener - except: - from urllib2 import build_opener - - class REPLApplication(ConsoleApplication): - def __init__(self): - self._script = None - self._seqno = 0 - self._ready = threading.Event() - self._stopping = threading.Event() - self._errors = 0 - config_dir = self._get_or_create_config_dir() - self._history = FileHistory(os.path.join(config_dir, 'history')) - self._completer = FridaCompleter(self) - self._cli = None - self._last_change_id = 0 - self._script_monitor = None - self._monitored_file = None - - super(REPLApplication, self).__init__(self._process_input, self._on_stop) - - self._dumb_stdin_reader = None if self._have_terminal else DumbStdinReader(valid_until=self._stopping) - - def _add_options(self, parser): - parser.add_option("-l", "--load", help="load SCRIPT", metavar="SCRIPT", - type='string', action='store', dest="user_script", default=None) - parser.add_option("-c", "--codeshare", help="load CODESHARE_URI", metavar="CODESHARE_URI", - type='string', action='store', dest="codeshare_uri", default=None) - parser.add_option("-e", "--eval", help="evaluate CODE", metavar="CODE", - type='string', action='append', dest="eval_items", default=None) - parser.add_option("-q", help="quiet mode (no prompt) and quit after -l and -e", - action='store_true', dest="quiet", default=False) - parser.add_option("--no-pause", help="automatically start main thread after startup", - action='store_true', dest="no_pause", default=False) - parser.add_option("-o", "--output", help="output to log file", dest="logfile", default=None) - - def _initialize(self, parser, options, args): - self._user_script = options.user_script - self._codeshare_uri = options.codeshare_uri - self._codeshare_script = None - self._pending_eval = options.eval_items - self._quiet = options.quiet - self._no_pause = options.no_pause - if options.logfile is not None: - try: - self._logfile = open(options.logfile, 'w') - except Exception as e: - self._update_status('Failed to open logfile "%s"' % options.logfile) - self._exit(1) - else: - self._logfile = None - - def _log(self, level, text): - ConsoleApplication._log(self, level, text) - if self._logfile is not None: - self._logfile.write(text + "\n") - - def _usage(self): - return "usage: %prog [options] target" - - def _needs_target(self): - return True - - def _start(self): - self._prompt_string = self._create_prompt() - - if self._codeshare_uri is not None: - self._codeshare_script = self._load_codeshare_script(self._codeshare_uri) - if self._codeshare_script is None: - self._print("Exiting!") - self._exit(1) - return - - try: - self._load_script() - except Exception as e: - self._update_status("Failed to load script: {error}".format(error=e)) - self._exit(1) - return - - if self._spawned_argv is not None: - if self._no_pause: - self._update_status("Spawned `{command}`. Resuming main thread!".format(command=" ".join(self._spawned_argv))) - self._do_magic("resume") - else: - self._update_status("Spawned `{command}`. Use %resume to let the main thread start executing!".format(command=" ".join(self._spawned_argv))) - else: - self._clear_status() - self._ready.set() - - def _on_stop(self): - def set_return(): - raise EOFError() - - self._stopping.set() - - if self._cli is not None: - try: - self._cli.eventloop.call_from_executor(set_return) - except Exception: - pass - - def _stop(self): - self._unload_script() - self._unmonitor_script() - - self._print("\nThank you for using Frida!") - - def _load_script(self): - self._monitor_script() - self._seqno += 1 - script = self._session.create_script(name="repl%d" % self._seqno, source=self._create_repl_script()) - script.set_log_handler(self._log) - self._unload_script() - self._script = script - def on_message(message, data): - self._reactor.schedule(lambda: self._process_message(message, data)) - script.on('message', on_message) - script.load() - - def _unload_script(self): - if self._script is None: - return - - try: - self._script.unload() - except: - pass - self._script = None - - def _monitor_script(self): - if self._monitored_file == self._user_script: - return - - self._unmonitor_script() - - if self._user_script is not None: - monitor = frida.FileMonitor(self._user_script) - monitor.on('change', self._on_change) - monitor.enable() - self._script_monitor = monitor - self._monitored_file = self._user_script - - def _unmonitor_script(self): - if self._script_monitor is None: - return - - self._script_monitor.disable() - self._script_monitor = None - - def _process_input(self, reactor): - if not self._quiet: - self._print_startup_message() - - while self._ready.wait(0.5) != True: - if not reactor.is_running(): - return - - while True: - expression = "" - line = "" - while len(expression) == 0 or line.endswith("\\"): - if not reactor.is_running(): - return - - prompt = "[%s]" % self._prompt_string + "-> " if len(expression) == 0 else "... " - - pending_eval = self._pending_eval - if pending_eval is not None: - if len(pending_eval) > 0: - expression = pending_eval.pop(0) - if not self._quiet: - self._print(prompt + expression) - else: - self._pending_eval = None - else: - if self._quiet: - self._exit_status = 0 if self._errors == 0 else 1 - return - - try: - if self._have_terminal: - # We create the prompt manually instead of using get_input, - # so we can use the cli in the _on_stop method - eventloop = create_eventloop() - - self._cli = CommandLineInterface( - application=create_prompt_application(prompt, history=self._history, completer=self._completer, lexer=JavascriptLexer), - eventloop=eventloop, - output=create_output()) - - try: - line = None - - document = self._cli.run() - - if document: - line = document.text - finally: - eventloop.close() - else: - line = self._dumb_stdin_reader.read_line(prompt) - self._print(line) - except EOFError: - if not self._have_terminal: - while not self._stopping.wait(1): - pass - return - except KeyboardInterrupt: - line = "" - if not self._have_terminal: - sys.stdout.write("\n" + prompt) - continue - if len(line.strip()) > 0: - if len(expression) > 0: - expression += "\n" - expression += line.rstrip("\\") - - if expression.endswith("?"): - try: - self._print_help(expression) - except JavaScriptError as e: - error = e.error - self._print(Fore.RED + Style.BRIGHT + error['name'] + Style.RESET_ALL + ": " + error['message']) - except frida.InvalidOperationError: - return - elif expression.startswith("%"): - self._do_magic(expression[1:].rstrip()) - elif expression in ("exit", "quit", "q"): - return - elif expression == "help": - self._print("Help: #TODO :)") - else: - if not self._eval_and_print(expression): - self._errors += 1 - - def _eval_and_print(self, expression): - success = False - try: - (t, value) = self._evaluate(expression) - if t in ('function', 'undefined', 'null'): - output = t - elif t == 'binary': - output = hexdump(value).rstrip("\n") - else: - output = json.dumps(value, sort_keys=True, indent=4, separators=(",", ": ")) - success = True - except JavaScriptError as e: - error = e.error - output = Fore.RED + Style.BRIGHT + error['name'] + Style.RESET_ALL + ": " + error['message'] - except frida.InvalidOperationError: - return success - self._print(output) - return success - - def _print_startup_message(self): - self._print("""\ - ____ - / _ | Frida {version} - A world-class dynamic instrumentation toolkit - | (_| | - > _ | Commands: - /_/ |_| help -> Displays the help system - . . . . object? -> Display information about 'object' - . . . . exit/quit -> Exit - . . . . - . . . . More info at http://www.frida.re/docs/home/""".format(version=frida.__version__)) - - def _print_help(self, expression): - # TODO: Figure out docstrings and implement here. This is real jankaty right now. - help_text = "" - if expression.endswith(".?"): - expression = expression[:-2] + "?" - - obj_to_identify = [x for x in expression.split(' ') if x.endswith("?")][0][:-1] - (obj_type, obj_value) = self._evaluate(obj_to_identify) - - if obj_type == "function": - signature = self._evaluate("%s.toString()" % obj_to_identify)[1] - clean_signature = signature.split("{")[0][:-1].split('function ')[-1] - - if "[native code]" in signature: - help_text += "Type: Function (native)\n" - else: - help_text += "Type: Function\n" - - help_text += "Signature: %s\n" % clean_signature - help_text += "Docstring: #TODO :)" - - elif obj_type == "object": - help_text += "Type: Object\n" - help_text += "Docstring: #TODO :)" - - elif obj_type == "boolean": - help_text += "Type: Boolean\n" - help_text += "Docstring: #TODO :)" - - elif obj_type == "string": - help_text += "Type: Boolean\n" - help_text += "Text: %s\n" % self._evaluate("%s.toString()" % obj_to_identify)[1] - help_text += "Docstring: #TODO :)" - - self._print(help_text) - - # Negative means at least abs(val) - 1 - _magic_command_args = { - 'resume': 0, - 'load': 1, - 'reload': 0, - 'unload': 0, - 'time': -2 # At least 1 arg - } - - def _do_magic(self, statement): - tokens = statement.split(" ") - command = tokens[0] - args = tokens[1:] - - required_args = self._magic_command_args.get(command) - - if required_args == None: - self._print("Unknown command: {}".format(command)) - self._print("Valid commands: {}".format(", ".join(self._magic_command_args.keys()))) - return - - atleast_args = False - if required_args < 0: - atleast_args = True - required_args = abs(required_args) - 1 - - if (not atleast_args and len(args) != required_args) or \ - (atleast_args and len(args) < required_args): - self._print("{cmd} command expects {atleast}{n} argument{s}".format( - cmd=command, atleast='atleast ' if atleast_args else '', n=required_args, s='' if required_args == 1 else ' ')) - return - - if command == 'resume': - self._reactor.schedule(lambda: self._resume()) - elif command == 'load': - old_user_script = self._user_script - self._user_script = os.path.abspath(args[0]) - if not self._reload(): - self._user_script = old_user_script - elif command == 'reload': - self._reload() - elif command == 'unload': - self._user_script = None - self._reload() - elif command == 'time': - self._eval_and_print(''' - (function () { - var _startTime = Date.now(); - var _result = eval({expression}); - var _endTime = Date.now(); - console.log('Time: ' + (_endTime - _startTime).toLocaleString() + ' ms.'); - return _result; - })();'''.format(expression=json.dumps(' '.join(args)))) - - def _reload(self): - completed = threading.Event() - result = [None] - def do_reload(): - try: - self._load_script() - except Exception as e: - result[0] = e - completed.set() - self._reactor.schedule(do_reload) - completed.wait() - if result[0] is None: - return True - else: - self._print("Failed to load script: {error}".format(error=result[0])) - return False - - def _create_prompt(self): - device_type = self._device.type - type_name = self._target[0] - if type_name == 'pid': - if self._target[1] == 0: - target = 'SystemSession' - else: - target = 'PID::%u' % self._target[1] - elif type_name == 'file': - target = os.path.basename(self._target[1][0]) - else: - target = self._target[1] - - if device_type in ('local', 'remote'): - prompt_string = "%s::%s" % (device_type.title(), target) - else: - prompt_string = "%s::%s" % (self._device.name, target) - - return prompt_string - - def _evaluate(self, text): - result = self._script.exports.evaluate(text) - if is_byte_array(result): - return ('binary', result) - elif isinstance(result, dict): - return ('binary', bytes()) - elif result[0] == 'error': - raise JavaScriptError(result[1]) - else: - return result - - def _process_message(self, message, data): - message_type = message['type'] - if message_type == 'error': - text = message.get('stack', message['description']) - self._log('error', text) - self._errors += 1 - else: - self._print("message:", message, "data:", data) - - def _on_change(self, changed_file, other_file, event_type): - if event_type == 'changes-done-hint': - return - self._last_change_id += 1 - change_id = self._last_change_id - self._reactor.schedule(lambda: self._process_change(change_id), delay=0.05) - - def _process_change(self, change_id): - if change_id != self._last_change_id: - return - try: - self._load_script() - except Exception as e: - self._print("Failed to load script: {error}".format(error=e)) - - def _create_repl_script(self): - user_script = "" - - if self._codeshare_script is not None: - user_script = self._codeshare_script - - if self._user_script is not None: - with codecs.open(self._user_script, 'rb', 'utf-8') as f: - user_script += f.read().rstrip("\r\n") + "\n\n// Frida REPL script:\n" - - return user_script + """\ - -rpc.exports.evaluate = function (expression) { - try { - var result = (1, eval)(expression); - if (result instanceof ArrayBuffer) { - return result; - } else { - var type = (result === null) ? 'null' : typeof result; - return [type, result]; - } - } catch (e) { - return ['error', { - name: e.name, - message: e.message, - stack: e.stack - }]; - } -}; -""" - - def _load_codeshare_script(self, uri): - trust_store = self._get_or_create_truststore() - - project_url = "https://codeshare.frida.re/api/project/{}/".format(uri) - response_json = None - try: - request = build_opener() - request.addheaders = [('User-Agent', 'Frida v{} | {}'.format(frida.__version__, platform.platform()))] - response = request.open(project_url) - response_content = response.read().decode('utf-8') - response_json = json.loads(response_content) - except Exception as e: - self._print("Got an unhandled exception while trying to retrieve {} - {}".format(uri, e)) - return None - - trusted_signature = trust_store.get(uri, "") - fingerprint = hashlib.sha256(response_json['source'].encode('utf-8')).hexdigest() - if fingerprint == trusted_signature: - return response_json['source'] - - self._print("""Hello! This is the first time you're running this particular snippet, or the snippet's source code has changed. - -Project Name: {project_name} -Author: {author} -Slug: {slug} -Fingerprint: {fingerprint} -URL: {url} - """.format( - project_name=response_json['project_name'], - author="@" + uri.split('/')[0], - slug=uri, - fingerprint=fingerprint, - url="https://codeshare.frida.re/@{}".format(uri) - )) - - while True: - prompt_string = "Are you sure you'd like to trust this project? [y/N] " - response = get_input(prompt_string) - - if response.lower() in ('n', 'no') or response == '': - return None - - if response.lower() in ('y', 'yes'): - self._print("Adding fingerprint {} to the trust store! You won't be prompted again unless the code changes.".format(fingerprint)) - script = response_json['source'] - self._update_truststore({ - uri: fingerprint - }) - return script - - def _get_or_create_config_dir(self): - config_dir = os.path.join(os.path.expanduser('~'), '.frida') - if not os.path.exists(config_dir): - os.makedirs(config_dir) - return config_dir - - def _update_truststore(self, record): - trust_store = self._get_or_create_truststore() - trust_store.update(record) - - config_dir = self._get_or_create_config_dir() - codeshare_trust_store = os.path.join(config_dir, "codeshare-truststore.json") - - with open(codeshare_trust_store, 'w') as f: - f.write(json.dumps(trust_store)) - - def _get_or_create_truststore(self): - config_dir = self._get_or_create_config_dir() - - codeshare_trust_store = os.path.join(config_dir, "codeshare-truststore.json") - - if os.path.exists(codeshare_trust_store): - try: - with open(codeshare_trust_store) as f: - trust_store = json.load(f) - except Exception as e: - self._print("Unable to load the codeshare truststore ({}), defaulting to an empty truststore. You will be prompted every time you want to run a script!".format(e)) - trust_store = {} - else: - with open(codeshare_trust_store, 'w') as f: - f.write(json.dumps({})) - trust_store = {} - - return trust_store - - class FridaCompleter(Completer): - def __init__(self, repl): - self._repl = repl - self._lexer = JavascriptLexer() - - def get_completions(self, document, complete_event): - prefix = document.text_before_cursor - - magic = len(prefix) > 0 and prefix[0] == '%' and not any(map(lambda c: c.isspace(), prefix)) - - tokens = list(self._lexer.get_tokens(prefix))[:-1] - - # 0.toString() is invalid syntax, - # but pygments doesn't seem to know that - for i in range(len(tokens) - 1): - if tokens[i][0] == Token.Literal.Number.Integer \ - and tokens[i + 1][0] == Token.Punctuation and tokens[i + 1][1] == '.': - tokens[i] = (Token.Literal.Number.Float, tokens[i][1] + tokens[i + 1][1]) - del tokens[i + 1] - - before_dot = '' - after_dot = '' - encountered_dot = False - for t in tokens[::-1]: - if t[0] in Token.Name.subtypes: - before_dot = t[1] + before_dot - elif t[0] == Token.Punctuation and t[1] == '.': - before_dot = '.' + before_dot - if not encountered_dot: - encountered_dot = True - after_dot = before_dot[1:] - before_dot = '' - else: - if encountered_dot: - # The value/contents of the string, number or array doesn't matter, - # so we just use the simplest value with that type - if t[0] in Token.Literal.String.subtypes: - before_dot = '""' + before_dot - elif t[0] in Token.Literal.Number.subtypes: - before_dot = '0.0' + before_dot - elif t[0] == Token.Punctuation and t[1] == ']': - before_dot = '[]' + before_dot - - break - - try: - if encountered_dot: - for key in self._get_keys("""try { - (function (o) { - "use strict"; - var k = Object.getOwnPropertyNames(o); - if (o !== null && o !== undefined) { - var p; - if (typeof o !== 'object') - p = o.__proto__; - else - p = Object.getPrototypeOf(o); - if (p !== null && p !== undefined) - k = k.concat(Object.getOwnPropertyNames(p)); - } - return k; - })(""" + before_dot + """); - } catch (e) { - []; - }"""): - if self._pattern_matches(after_dot, key): - yield Completion(key, -len(after_dot)) - else: - if magic: - keys = self._repl._magic_command_args.keys() - else: - keys = self._get_keys("Object.getOwnPropertyNames(this)") - for key in keys: - if not self._pattern_matches(before_dot, key) or (key.startswith('_') and before_dot == ''): - continue - yield Completion(key, -len(before_dot)) - except frida.InvalidOperationError: - pass - except Exception as e: - self._repl._print(e) - - def _get_keys(self, code): - return sorted( - filter(self._is_valid_name, - set(self._repl._evaluate(code)[1]))) - - def _is_valid_name(self, name): - tokens = list(self._lexer.get_tokens(name)) - return len(tokens) == 2 and tokens[0][0] in Token.Name.subtypes - - def _pattern_matches(self, pattern, text): - return re.search(re.escape(pattern), text, re.IGNORECASE) != None - - def hexdump(src, length=16): - try: - xrange - except NameError: - xrange = range - FILTER = "".join([(len(repr(chr(x))) == 3) and chr(x) or "." for x in range(256)]) - lines = [] - for c in xrange(0, len(src), length): - chars = src[c:c + length] - hex = " ".join(["%02x" % x for x in iterbytes(chars)]) - printable = ''.join(["%s" % ((x <= 127 and FILTER[x]) or ".") for x in iterbytes(chars)]) - lines.append("%04x %-*s %s\n" % (c, length * 3, hex, printable)) - return "".join(lines) - - def is_byte_array(value): - if sys.version_info[0] >= 3: - return isinstance(value, bytes) - else: - return isinstance(value, str) - - if sys.version_info[0] >= 3: - iterbytes = lambda x: iter(x) - else: - def iterbytes(data): - return (ord(char) for char in data) - - app = REPLApplication() - app.run() - -class JavaScriptError(Exception): - def __init__(self, error): - super(JavaScriptError, self).__init__(error['message']) - - self.error = error - -class DumbStdinReader(object): - def __init__(self, valid_until): - self._valid_until = valid_until - - self._prompt = None - self._result = None - self._lock = threading.Lock() - self._cond = threading.Condition(self._lock) - - worker = threading.Thread(target=self._process_requests, name="stdin-reader") - worker.daemon = True - worker.start() - - signal.signal(signal.SIGINT, lambda n, f: self._cancel_line()) - - def read_line(self, prompt_string): - with self._lock: - self._prompt = prompt_string - self._cond.notify() - - with self._lock: - while self._result is None: - if self._valid_until.is_set(): - raise EOFError() - self._cond.wait(1) - line, error = self._result - self._result = None - - if error is not None: - raise error - - return line - - def _process_requests(self): - error = None - while error is None: - with self._lock: - while self._prompt is None: - self._cond.wait() - prompt = self._prompt - - try: - line = get_input(prompt) - except Exception as e: - line = None - error = e - - with self._lock: - self._prompt = None - self._result = (line, error) - self._cond.notify() - - def _cancel_line(self): - with self._lock: - self._prompt = None - self._result = (None, KeyboardInterrupt()) - self._cond.notify() - -try: - input_impl = raw_input -except NameError: - input_impl = input - -def get_input(prompt_string): - return input_impl(prompt_string) - - -if __name__ == '__main__': - main() diff --git a/src/frida/tracer.py b/src/frida/tracer.py deleted file mode 100755 index 602fc96..0000000 --- a/src/frida/tracer.py +++ /dev/null @@ -1,923 +0,0 @@ -# -*- coding: utf-8 -*- -from __future__ import unicode_literals, print_function - -import binascii -import codecs -import os -import platform -import re -import subprocess -import threading -import time - -from frida import FileMonitor -from frida.core import Function, Module, ModuleFunction, ObjCMethod - - -class TracerProfileBuilder(object): - _RE_REL_ADDRESS = re.compile("(?P[^\s!]+)!(?P(0x)?[0-9a-fA-F]+)") - - def __init__(self): - self._spec = [] - - def include_modules(self, *module_name_globs): - for m in module_name_globs: - self._spec.append(('include', 'module', m)) - return self - - def exclude_modules(self, *module_name_globs): - for m in module_name_globs: - self._spec.append(('exclude', 'module', m)) - return self - - def include(self, *function_name_globs): - for f in function_name_globs: - self._spec.append(('include', 'function', f)) - return self - - def exclude(self, *function_name_globs): - for f in function_name_globs: - self._spec.append(('exclude', 'function', f)) - return self - - def include_relative_address(self, *address_rel_offsets): - for f in address_rel_offsets: - m = TracerProfileBuilder._RE_REL_ADDRESS.search(f) - if m is None: - continue - self._spec.append(('include', 'relative_function', { - 'module': m.group('module'), - 'offset': int(m.group('offset'), base=16) - })) - return self - - def include_imports(self, *module_name_globs): - for m in module_name_globs: - self._spec.append(('include', 'imports', m)) - return self - - def include_objc_method(self, *function_name_globs): - for f in function_name_globs: - self._spec.append(('include', 'objc_method', f)) - return self - - def include_debug_symbol(self, *function_name_globs): - for f in function_name_globs: - self._spec.append(('include', 'debug_symbol', f)) - return self - - def build(self): - return TracerProfile(self._spec) - -class TracerProfile(object): - _BLACKLIST = set([ - "libSystem.B.dylib!dyld_stub_binder" - ]) - - def __init__(self, spec): - self._spec = spec - - def resolve(self, session, log_handler=None): - script = session.create_script(name="profile-resolver", source=self._create_resolver_script()) - script.set_log_handler(log_handler) - def on_message(message, data): - print(message) - script.on('message', on_message) - script.load() - try: - data = script.exports.resolve(self._spec) - finally: - script.unload() - - modules = {} - for module_id, m in data['modules'].items(): - module = Module(m['name'], int(m['base'], 16), m['size'], m['path'], session) - modules[int(module_id)] = module - - working_set = [] - for target in data['targets']: - objc = target.get('objc') - if objc is not None: - method = objc['method'] - of = ObjCMethod(method['type'], objc['className'], method['name'], int(target['address'], 16)) - working_set.append(of) - else: - name = target['name'] - absolute_address = int(target['address'], 16) - module_id = target.get('module') - if module_id is not None: - module = modules[module_id] - relative_address = absolute_address - module.base_address - exported = not target.get('private', False) - mf = ModuleFunction(module, name, relative_address, exported) - if not self._is_blacklisted(mf): - working_set.append(mf) - else: - f = Function(name, absolute_address) - working_set.append(f) - return working_set - - def _is_blacklisted(self, module_function): - key = module_function.module.name + "!" + module_function.name - return key in TracerProfile._BLACKLIST - - def _create_resolver_script(self): - return r""""use strict"; - -rpc.exports = { - resolve: function (spec) { - var workingSet = spec.reduce(function (workingSet, item) { - var operation = item[0]; - var scope = item[1]; - var param = item[2]; - switch (scope) { - case 'module': - if (operation === 'include') - workingSet = includeModule(param, workingSet); - else if (operation === 'exclude') - workingSet = excludeModule(param, workingSet); - break; - case 'function': - if (operation === 'include') - workingSet = includeFunction(param, workingSet); - else if (operation === 'exclude') - workingSet = excludeFunction(param, workingSet); - break; - case 'relative_function': - if (operation === 'include') - workingSet = includeRelativeFunction(param, workingSet); - break; - case 'imports': - if (operation === 'include') - workingSet = includeImports(param, workingSet); - break; - case 'objc_method': - if (operation === 'include') - workingSet = includeObjCMethod(param, workingSet); - break; - case 'debug_symbol': - if (operation === 'include') - workingSet = includeDebugSymbol(param, workingSet); - break; - } - return workingSet; - }, {}); - - var modules = {}; - var targets = []; - for (var address in workingSet) { - if (workingSet.hasOwnProperty(address)) { - var target = workingSet[address]; - var moduleId = target.module; - if (moduleId !== undefined && !modules.hasOwnProperty(moduleId)) { - var m = allModules()[moduleId]; - delete m._cachedFunctionExports; - modules[moduleId] = m; - } - targets.push(target); - } - } - return { - modules: modules, - targets: targets - }; - } -}; - -function includeModule(pattern, workingSet) { - moduleResolver().enumerateMatchesSync('exports:' + pattern + '!*').forEach(function (m) { - workingSet[m.address.toString()] = moduleExportFromMatch(m); - }); - return workingSet; -} - -function excludeModule(pattern, workingSet) { - moduleResolver().enumerateMatchesSync('exports:' + pattern + '!*').forEach(function (m) { - delete workingSet[m.address.toString()]; - }); - return workingSet; -} - -function includeFunction(pattern, workingSet) { - moduleResolver().enumerateMatchesSync('exports:*!' + pattern).forEach(function (m) { - workingSet[m.address.toString()] = moduleExportFromMatch(m); - }); - return workingSet; -} - -function excludeFunction(pattern, workingSet) { - moduleResolver().enumerateMatchesSync('exports:*!' + pattern).forEach(function (m) { - delete workingSet[m.address.toString()]; - }); - return workingSet; -} - -function includeRelativeFunction(func, workingSet) { - var relativeToModule = func.module; - var modules = allModules(); - for (var moduleIndex = 0; moduleIndex !== modules.length; moduleIndex++) { - var module = modules[moduleIndex]; - if (module.path === relativeToModule || module.name === relativeToModule) { - var relativeAddress = ptr(func.offset); - var absoluteAddress = module.base.add(relativeAddress); - workingSet[absoluteAddress] = { - name: "sub_" + relativeAddress.toString(16), - address: absoluteAddress, - module: moduleIndex, - private: true - }; - } - } - return workingSet; -} - -function includeImports(pattern, workingSet) { - var matches; - if (pattern === null) { - var mainModule = allModules()[0].path; - matches = moduleResolver().enumerateMatchesSync('imports:' + mainModule + '!*'); - } else { - matches = moduleResolver().enumerateMatchesSync('imports:' + pattern + '!*'); - } - - matches.map(moduleExportFromMatch).forEach(function (e) { - workingSet[e.address.toString()] = e; - }); - - return workingSet; -} - -function includeObjCMethod(pattern, workingSet) { - objcResolver().enumerateMatchesSync(pattern).forEach(function (m) { - workingSet[m.address.toString()] = objcMethodFromMatch(m); - }); - return workingSet; -} - -function includeDebugSymbol(pattern, workingSet) { - DebugSymbol.findFunctionsMatching(pattern).forEach(function (addr) { - workingSet[addr.toString()] = debugSymbolFromAddress(addr); - }); - return workingSet; -} - -var cachedModuleResolver = null; -function moduleResolver() { - if (cachedModuleResolver === null) - cachedModuleResolver = new ApiResolver('module'); - return cachedModuleResolver; -} - -var cachedObjcResolver = null; -function objcResolver() { - if (cachedObjcResolver === null) { - try { - cachedObjcResolver = new ApiResolver('objc'); - } catch (e) { - throw new Error("Objective-C runtime is not available"); - } - } - return cachedObjcResolver; -} - -var cachedModules = null; -function allModules() { - if (cachedModules === null) { - cachedModules = Process.enumerateModulesSync(); - cachedModules._idByPath = cachedModules.reduce(function (mappings, module, index) { - mappings[module.path] = index; - return mappings; - }, {}); - cachedModules._idByName = cachedModules.reduce(function (mappings, module, index) { - mappings[module.name] = index; - return mappings; - }, {}); - } - return cachedModules; -} - -function moduleExportFromMatch(m) { - var encodedName = m.name; - var delimiterIndex = encodedName.indexOf('!'); - var modulePath = encodedName.substring(0, delimiterIndex); - var functionName = encodedName.substring(delimiterIndex + 1); - return { - name: functionName, - address: m.address, - module: allModules()._idByPath[modulePath] - }; -} - -function objcMethodFromMatch(m) { - var encodedName = m.name; - var methodType = encodedName[0]; - var delimiterIndex = encodedName.indexOf(' ', 3); - var className = encodedName.substring(2, delimiterIndex); - var methodName = encodedName.substring(delimiterIndex + 1, encodedName.length - 1); - return { - objc: { - className: className, - method: { - type: methodType, - name: methodName - } - }, - address: m.address - }; -} - -function debugSymbolFromAddress(address) { - var symbol = DebugSymbol.fromAddress(address); - return { - name: symbol.name, - address: symbol.address, - module: allModules()._idByName[symbol.moduleName] - }; -} -""" - -class Tracer(object): - def __init__(self, reactor, repository, profile, log_handler=None): - self._reactor = reactor - self._repository = repository - self._profile = profile - self._script = None - self._log_handler = log_handler - - def start_trace(self, session, ui): - def on_create(*args): - ui.on_trace_handler_create(*args) - self._repository.on_create(on_create) - - def on_load(*args): - ui.on_trace_handler_load(*args) - self._repository.on_load(on_load) - - def on_update(function, handler, source): - self._script.exports.update([{ - 'name': function.name, - 'absolute_address': hex(function.absolute_address), - 'handler': handler - }]) - self._repository.on_update(on_update) - - def on_message(message, data): - self._reactor.schedule(lambda: self._process_message(message, data, ui)) - - ui.on_trace_progress('resolve') - working_set = self._profile.resolve(session, log_handler=self._log_handler) - ui.on_trace_progress('instrument') - self._script = session.create_script(name="tracer", source=self._create_trace_script()) - self._script.set_log_handler(self._log_handler) - self._script.on('message', on_message) - self._script.load() - for chunk in [working_set[i:i+1000] for i in range(0, len(working_set), 1000)]: - targets = [{ - 'name': function.name, - 'absolute_address': hex(function.absolute_address), - 'handler': self._repository.ensure_handler(function) - } for function in chunk] - self._script.exports.add(targets) - - self._repository.commit_handlers() - - self._reactor.schedule(lambda: ui.on_trace_progress('ready')) - - return working_set - - def stop(self): - if self._script is not None: - try: - self._script.unload() - except: - pass - self._script = None - - def _create_trace_script(self): - return """"use strict"; - -var started = Date.now(); -var handlers = {}; -var state = {}; -var pending = []; -var timer = null; - -rpc.exports = { - dispose: function () { - flush(); - }, - add: function (targets) { - targets.forEach(function (target) { - var h = [parseHandler(target)]; - var name = target.name; - var targetAddress = target.absolute_address; - target = null; - - handlers[targetAddress] = h; - - function invokeCallback(callback, context, param) { - if (callback === undefined) - return; - - var timestamp = Date.now() - started; - var threadId = context.threadId; - var depth = context.depth; - - function log(message) { - emit([timestamp, threadId, depth, targetAddress, message]); - } - - callback.call(context, log, param, state); - } - - try { - Interceptor.attach(ptr(targetAddress), { - onEnter: function (args) { - invokeCallback(h[0].onEnter, this, args); - }, - onLeave: function (retval) { - invokeCallback(h[0].onLeave, this, retval); - } - }); - } catch (e) { - send({ - from: "/targets", - name: '+error', - payload: { - message: "Skipping '" + name + "': " + e.message - } - }); - } - }); - }, - update: function (targets) { - targets.forEach(function (target) { - handlers[target.absolute_address][0] = parseHandler(target); - }); - } -}; - -function emit(event) { - pending.push(event); - - if (timer === null) - timer = setTimeout(flush, 50); -} - -function flush() { - if (timer !== null) { - clearTimeout(timer); - timer = null; - } - - if (pending.length === 0) - return; - - var items = pending; - pending = []; - - send({ - from: "/events", - name: '+add', - payload: { - items: items - } - }); -} - -function parseHandler(target) { - try { - return (1, eval)("(" + target.handler + ")"); - } catch (e) { - send({ - from: "/targets", - name: '+error', - payload: { - message: "Invalid handler for '" + target.name + "': " + e.message - } - }); - return {}; - } -} -""" - - def _process_message(self, message, data, ui): - handled = False - if message['type'] == 'send': - stanza = message['payload'] - if isinstance(stanza, dict): - if stanza['from'] == "/events": - if stanza['name'] == '+add': - events = [(timestamp, thread_id, depth, int(target_address.rstrip("L"), 16), message) for timestamp, thread_id, depth, target_address, message in stanza['payload']['items']] - ui.on_trace_events(events) - handled = True - elif stanza['from'] == "/targets" and stanza['name'] == '+error': - ui.on_trace_error(stanza['payload']) - handled = True - if not handled: - print(message) - -class Repository(object): - def __init__(self): - self._on_create_callback = None - self._on_load_callback = None - self._on_update_callback = None - - def ensure_handler(self, function): - raise NotImplementedError("not implemented") - - def commit_handlers(self): - pass - - def on_create(self, callback): - self._on_create_callback = callback - - def on_load(self, callback): - self._on_load_callback = callback - - def on_update(self, callback): - self._on_update_callback = callback - - def _notify_create(self, function, handler, source): - if self._on_create_callback is not None: - self._on_create_callback(function, handler, source) - - def _notify_load(self, function, handler, source): - if self._on_load_callback is not None: - self._on_load_callback(function, handler, source) - - def _notify_update(self, function, handler, source): - if self._on_update_callback is not None: - self._on_update_callback(function, handler, source) - - def _create_stub_handler(self, function): - if isinstance(function, ObjCMethod): - display_name = function.display_name() - state = {"index": 2} - def objc_arg(m): - index = state["index"] - r = ":\" + args[%d] + \" " % index - state["index"] = index + 1 - return r - - log_str = '"' + re.sub(r':', objc_arg, display_name) + '"' - if log_str.endswith("\" ]\""): - log_str = log_str[:-3] + "]\"" - else: - display_name = function.name - - for man_section in (2, 3): - args = [] - try: - with open(os.devnull, 'w') as devnull: - man_argv = ["man"] - if platform.system() != "Darwin": - man_argv.extend(["-E", "UTF-8"]) - man_argv.extend(["-P", "col -b", str(man_section), function.name]) - output = subprocess.check_output(man_argv, stderr=devnull) - match = re.search(r"^SYNOPSIS(?:.|\n)*?((?:^.+$\n)* {5}\w+[ \*\n]*" + function.name + r"\((?:.+\,\s*?$\n)*?(?:.+\;$\n))(?:.|\n)*^DESCRIPTION", output.decode('UTF-8', errors='replace'), re.MULTILINE) - if match: - decl = match.group(1) - - for argm in re.finditer(r"[\(,]\s*(.+?)\s*\b(\w+)(?=[,\)])", decl): - typ = argm.group(1) - arg = argm.group(2) - if arg == "void": - continue - if arg == "...": - args.append("\", ...\" +"); - continue - - cast_pre = "" - cast_post = "" - annotate_pre = "" - annotate_post = "" - - normalized_type = re.sub(r"\s+", "", typ) - if normalized_type.endswith("*restrict"): - normalized_type = normalized_type[:-8] - if normalized_type in ("char*", "constchar*"): - cast_pre = "Memory.readUtf8String(" - cast_post = ")" - annotate_pre = "\\\"" - annotate_post = " + \"\\\"\"" - - arg_index = len(args) - - args.append("\"%(arg_delimiter)s%(arg_name)s=%(annotate_pre)s\" + %(cast_pre)sargs[%(arg_index)s]%(cast_post)s%(annotate_post)s +" % { - "arg_name": arg, - "arg_index": arg_index, - "arg_delimiter": ", " if arg_index > 0 else "", - "cast_pre": cast_pre, - "cast_post": cast_post, - "annotate_pre": annotate_pre, - "annotate_post": annotate_post - }) - break - except Exception as e: - pass - - if len(args) == 0: - log_str = "\"%(name)s()\"" % { "name": function.name } - else: - indent_outer = " " - indent_inner = " " - log_str = "\"%(name)s(\" +\n%(indent_inner)s%(args)s\n%(indent_outer)s\")\"" % { - "name": function.name, - "args": ("\n" + indent_inner).join(args), - "indent_outer": indent_outer, - "indent_inner": indent_inner - } - - return """\ -/* - * Auto-generated by Frida. Please modify to match the signature of %(display_name)s. - * This stub is currently auto-generated from manpages when available. - * - * For full API reference, see: http://www.frida.re/docs/javascript-api/ - */ - -{ - /** - * Called synchronously when about to call %(display_name)s. - * - * @this {object} - Object allowing you to store state for use in onLeave. - * @param {function} log - Call this function with a string to be presented to the user. - * @param {array} args - Function arguments represented as an array of NativePointer objects. - * For example use Memory.readUtf8String(args[0]) if the first argument is a pointer to a C string encoded as UTF-8. - * It is also possible to modify arguments by assigning a NativePointer object to an element of this array. - * @param {object} state - Object allowing you to keep state across function calls. - * Only one JavaScript function will execute at a time, so do not worry about race-conditions. - * However, do not use this to store function arguments across onEnter/onLeave, but instead - * use "this" which is an object for keeping state local to an invocation. - */ - onEnter: function (log, args, state) { - log(%(log_str)s); - }, - - /** - * Called synchronously when about to return from %(display_name)s. - * - * See onEnter for details. - * - * @this {object} - Object allowing you to access state stored in onEnter. - * @param {function} log - Call this function with a string to be presented to the user. - * @param {NativePointer} retval - Return value represented as a NativePointer object. - * @param {object} state - Object allowing you to keep state across function calls. - */ - onLeave: function (log, retval, state) { - } -} -""" % {"display_name": display_name, "log_str": log_str} - -class MemoryRepository(Repository): - def __init__(self): - super(MemoryRepository, self).__init__() - self._handlers = {} - - def ensure_handler(self, function): - handler = self._handlers.get(function) - if handler is None: - handler = self._create_stub_handler(function) - self._handlers[function] = handler - self._notify_create(function, handler, "memory") - else: - self._notify_load(function, handler, "memory") - return handler - -class FileRepository(Repository): - def __init__(self, reactor): - super(FileRepository, self).__init__() - self._reactor = reactor - self._handler_by_address = {} - self._handler_by_file = {} - self._changed_files = set() - self._last_change_id = 0 - self._repo_dir = os.path.join(os.getcwd(), "__handlers__") - self._repo_monitors = {} - - def ensure_handler(self, function): - entry = self._handler_by_address.get(function.absolute_address) - if entry is not None: - (function, handler, handler_file) = entry - return handler - - handler = None - handler_files_to_try = [] - - if isinstance(function, ModuleFunction): - module_dir = os.path.join(self._repo_dir, to_filename(function.module.name)) - module_handler_file = os.path.join(module_dir, to_handler_filename(function.name)) - handler_files_to_try.append(module_handler_file) - - any_module_handler_file = os.path.join(self._repo_dir, to_handler_filename(function.name)) - handler_files_to_try.append(any_module_handler_file) - - for handler_file in handler_files_to_try: - if os.path.isfile(handler_file): - with codecs.open(handler_file, 'r', 'utf-8') as f: - handler = f.read() - self._notify_load(function, handler, handler_file) - break - - if handler is None: - handler = self._create_stub_handler(function) - handler_file = handler_files_to_try[0] - handler_dir = os.path.dirname(handler_file) - if not os.path.isdir(handler_dir): - os.makedirs(handler_dir) - with open(handler_file, 'w') as f: - f.write(handler) - self._notify_create(function, handler, handler_file) - - entry = (function, handler, handler_file) - self._handler_by_address[function.absolute_address] = entry - self._handler_by_file[handler_file] = entry - - self._ensure_monitor(handler_file) - - return handler - - def _ensure_monitor(self, handler_file): - handler_dir = os.path.dirname(handler_file) - monitor = self._repo_monitors.get(handler_dir) - if monitor is None: - monitor = FileMonitor(handler_dir) - monitor.on('change', self._on_change) - self._repo_monitors[handler_dir] = monitor - - def commit_handlers(self): - for monitor in self._repo_monitors.values(): - monitor.enable() - - def _on_change(self, changed_file, other_file, event_type): - if changed_file not in self._handler_by_file or event_type == 'changes-done-hint': - return - self._changed_files.add(changed_file) - self._last_change_id += 1 - change_id = self._last_change_id - self._reactor.schedule(lambda: self._sync_handlers(change_id), delay=0.05) - - def _sync_handlers(self, change_id): - if change_id != self._last_change_id: - return - changes = self._changed_files.copy() - self._changed_files.clear() - for changed_handler_file in changes: - (function, old_handler, handler_file) = self._handler_by_file[changed_handler_file] - with codecs.open(handler_file, 'r', 'utf-8') as f: - new_handler = f.read() - changed = new_handler != old_handler - if changed: - entry = (function, new_handler, handler_file) - self._handler_by_address[function.absolute_address] = entry - self._handler_by_file[handler_file] = entry - self._notify_update(function, new_handler, handler_file) - -class UI(object): - def on_trace_progress(self, operation): - pass - - def on_trace_error(self, error): - pass - - def on_trace_events(self, events): - pass - - def on_trace_handler_create(self, function, handler, source): - pass - - def on_trace_handler_load(self, function, handler, source): - pass - - -def main(): - from colorama import Fore, Style - from frida.application import ConsoleApplication, input_with_timeout - - class TracerApplication(ConsoleApplication, UI): - def __init__(self): - super(TracerApplication, self).__init__(self._await_ctrl_c) - self._palette = [Fore.CYAN, Fore.MAGENTA, Fore.YELLOW, Fore.GREEN, Fore.RED, Fore.BLUE] - self._next_color = 0 - self._attributes_by_thread_id = {} - self._last_event_tid = -1 - - def _add_options(self, parser): - pb = TracerProfileBuilder() - def process_builder_arg(option, opt_str, value, parser, method, **kwargs): - method(value) - parser.add_option("-I", "--include-module", help="include MODULE", metavar="MODULE", - type='string', action='callback', callback=process_builder_arg, callback_args=(pb.include_modules,)) - parser.add_option("-X", "--exclude-module", help="exclude MODULE", metavar="MODULE", - type='string', action='callback', callback=process_builder_arg, callback_args=(pb.exclude_modules,)) - parser.add_option("-i", "--include", help="include FUNCTION", metavar="FUNCTION", - type='string', action='callback', callback=process_builder_arg, callback_args=(pb.include,)) - parser.add_option("-x", "--exclude", help="exclude FUNCTION", metavar="FUNCTION", - type='string', action='callback', callback=process_builder_arg, callback_args=(pb.exclude,)) - parser.add_option("-a", "--add", help="add MODULE!OFFSET", metavar="MODULE!OFFSET", - type='string', action='callback', callback=process_builder_arg, callback_args=(pb.include_relative_address,)) - parser.add_option("-T", "--include-imports", help="include program's imports", - action='callback', callback=process_builder_arg, callback_args=(pb.include_imports,)) - parser.add_option("-t", "--include-module-imports", help="include MODULE imports", metavar="MODULE", - type='string', action='callback', callback=process_builder_arg, callback_args=(pb.include_imports,)) - parser.add_option("-m", "--include-objc-method", help="include OBJC_METHOD", metavar="OBJC_METHOD", - type='string', action='callback', callback=process_builder_arg, callback_args=(pb.include_objc_method,)) - parser.add_option("-s", "--include-debug-symbol", help="include DEBUG_SYMBOL", metavar="DEBUG_SYMBOL", - type='string', action='callback', callback=process_builder_arg, callback_args=(pb.include_debug_symbol,)) - self._profile_builder = pb - - def _usage(self): - return "usage: %prog [options] target" - - def _initialize(self, parser, options, args): - self._tracer = None - self._targets = None - self._profile = self._profile_builder.build() - - def _needs_target(self): - return True - - def _start(self): - self._tracer = Tracer(self._reactor, FileRepository(self._reactor), self._profile, log_handler=self._log) - try: - self._targets = self._tracer.start_trace(self._session, self) - except Exception as e: - self._update_status("Failed to start tracing: {error}".format(error=e)) - self._exit(1) - - def _stop(self): - self._tracer.stop() - self._tracer = None - - def _await_ctrl_c(self, reactor): - while reactor.is_running(): - try: - input_with_timeout(0.5) - except KeyboardInterrupt: - break - - def on_trace_progress(self, operation): - if operation == 'resolve': - self._update_status("Resolving functions...") - elif operation == 'instrument': - self._update_status("Instrumenting functions...") - elif operation == 'ready': - if len(self._targets) == 1: - plural = "" - else: - plural = "s" - self._update_status("Started tracing %d function%s. Press Ctrl+C to stop." % (len(self._targets), plural)) - self._resume() - - def on_trace_error(self, error): - self._print(Fore.RED + Style.BRIGHT + "Error" + Style.RESET_ALL + ": " + error['message']) - - def on_trace_events(self, events): - no_attributes = Style.RESET_ALL - for timestamp, thread_id, depth, target_address, message in events: - indent = depth * " | " - attributes = self._get_attributes(thread_id) - if thread_id != self._last_event_tid: - self._print("%s /* TID 0x%x */%s" % (attributes, thread_id, Style.RESET_ALL)) - self._last_event_tid = thread_id - self._print("%6d ms %s%s%s%s" % (timestamp, attributes, indent, message, no_attributes)) - - def on_trace_handler_create(self, function, handler, source): - self._print("%s: Auto-generated handler at \"%s\"" % (function, source.replace("\\", "\\\\"))) - - def on_trace_handler_load(self, function, handler, source): - self._print("%s: Loaded handler at \"%s\"" % (function, source.replace("\\", "\\\\"))) - - def _get_attributes(self, thread_id): - attributes = self._attributes_by_thread_id.get(thread_id, None) - if attributes is None: - color = self._next_color - self._next_color += 1 - attributes = self._palette[color % len(self._palette)] - if (1 + int(color / len(self._palette))) % 2 == 0: - attributes += Style.BRIGHT - self._attributes_by_thread_id[thread_id] = attributes - return attributes - - app = TracerApplication() - app.run() - -def to_filename(name): - result = "" - for c in name: - if c.isalnum() or c == ".": - result += c - else: - result += "_" - return result - -def to_handler_filename(name): - full_filename = to_filename(name) - if len(full_filename) <= 41: - return full_filename + ".js" - crc = binascii.crc32(full_filename.encode()) - return full_filename[0:32] + "_%08x.js" % crc - -if __name__ == '__main__': - main() diff --git a/src/meson.build b/src/meson.build index 54ec7e2..54d550a 100644 --- a/src/meson.build +++ b/src/meson.build @@ -1,5 +1,4 @@ subdir('frida') -subdir('scripts') extra_link_args = [] if host_os_family == 'darwin' diff --git a/src/scripts/frida-discover.in b/src/scripts/frida-discover.in deleted file mode 100644 index 18c4507..0000000 --- a/src/scripts/frida-discover.in +++ /dev/null @@ -1,11 +0,0 @@ -#!@PYTHON@ - -from os.path import dirname, join, realpath -import sys -sys.path.insert(1, join(dirname(dirname(realpath(__file__))), '@pythondir@')) - -import frida -import frida.discoverer - -if __name__ == '__main__': - frida.discoverer.main() diff --git a/src/scripts/frida-kill.in b/src/scripts/frida-kill.in deleted file mode 100644 index 133fce1..0000000 --- a/src/scripts/frida-kill.in +++ /dev/null @@ -1,10 +0,0 @@ -#!@PYTHON@ - -from os.path import dirname, join, realpath -import sys -sys.path.insert(1, join(dirname(dirname(realpath(__file__))), '@pythondir@')) - -import frida.kill - -if __name__ == '__main__': - frida.kill.main() diff --git a/src/scripts/frida-ls-devices.in b/src/scripts/frida-ls-devices.in deleted file mode 100644 index adfc2c9..0000000 --- a/src/scripts/frida-ls-devices.in +++ /dev/null @@ -1,10 +0,0 @@ -#!@PYTHON@ - -from os.path import dirname, join, realpath -import sys -sys.path.insert(1, join(dirname(dirname(realpath(__file__))), '@pythondir@')) - -import frida.lsd - -if __name__ == '__main__': - frida.lsd.main() diff --git a/src/scripts/frida-ps.in b/src/scripts/frida-ps.in deleted file mode 100644 index 84b424a..0000000 --- a/src/scripts/frida-ps.in +++ /dev/null @@ -1,10 +0,0 @@ -#!@PYTHON@ - -from os.path import dirname, join, realpath -import sys -sys.path.insert(1, join(dirname(dirname(realpath(__file__))), '@pythondir@')) - -import frida.ps - -if __name__ == '__main__': - frida.ps.main() diff --git a/src/scripts/frida-trace.in b/src/scripts/frida-trace.in deleted file mode 100644 index bda25ff..0000000 --- a/src/scripts/frida-trace.in +++ /dev/null @@ -1,11 +0,0 @@ -#!@PYTHON@ - -from os.path import dirname, join, realpath -import sys -sys.path.insert(1, join(dirname(dirname(realpath(__file__))), '@pythondir@')) - -import frida -import frida.tracer - -if __name__ == '__main__': - frida.tracer.main() diff --git a/src/scripts/frida.in b/src/scripts/frida.in deleted file mode 100644 index e269e00..0000000 --- a/src/scripts/frida.in +++ /dev/null @@ -1,10 +0,0 @@ -#!@PYTHON@ - -from os.path import dirname, join, realpath -import sys -sys.path.insert(1, join(dirname(dirname(realpath(__file__))), '@pythondir@')) - -import frida.repl - -if __name__ == '__main__': - frida.repl.main() diff --git a/src/scripts/meson.build b/src/scripts/meson.build deleted file mode 100644 index ffea21a..0000000 --- a/src/scripts/meson.build +++ /dev/null @@ -1,24 +0,0 @@ -scripts = [ - 'frida', - 'frida-discover', - 'frida-kill', - 'frida-ls-devices', - 'frida-ps', - 'frida-trace', -] - -cdata = configuration_data() -cdata.set('PYTHON', python) -cdata.set('pythondir', python_site_packages) - -foreach s : scripts - generated_script = configure_file( - input: s + '.in', - output: s, - configuration: cdata, - ) - install_data(generated_script, - install_dir: get_option('bindir'), - install_mode: 'rwxr-xr-x', - ) -endforeach diff --git a/tests/__init__.py b/tests/__init__.py index dced6aa..d900320 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,5 +1,4 @@ # -*- coding: utf-8 -*- from .test_core import TestCore -from .test_discoverer import TestDiscoverer -from .test_tracer import TestTracer +from .test_rpc import TestRpc diff --git a/tests/test_core.py b/tests/test_core.py index 52234a8..80e8ab0 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -1,10 +1,5 @@ # -*- coding: utf-8 -*- -from .data import target_program -import platform -import subprocess -import sys -import threading try: import unittest2 as unittest except: @@ -14,94 +9,10 @@ import frida class TestCore(unittest.TestCase): - @classmethod - def setUpClass(cls): - system = platform.system() - cls.target = subprocess.Popen([target_program], stdin=subprocess.PIPE) - cls.session = frida.attach(cls.target.pid) - - @classmethod - def tearDownClass(cls): - cls.session.detach() - cls.target.terminate() - cls.target.stdin.close() - def test_enumerate_devices(self): devices = frida.get_device_manager().enumerate_devices() self.assertTrue(len(devices) > 0) - def test_enumerate_modules(self): - modules = self.session.enumerate_modules() - self.assertGreater(len(modules), 1) - m = modules[0] - self.assertIsInstance(repr(m), str) - self.assertIsInstance(str(m), str) - - def test_enumerate_ranges(self): - ranges = self.session.enumerate_ranges('r--') - self.assertTrue(len(ranges) > 0) - r = ranges[0] - self.assertIsInstance(repr(r), str) - self.assertIsInstance(str(r), str) - - def test_find_base_address(self): - m = self.session.enumerate_modules()[0] - self.assertEqual(self.session.find_base_address(m.name), m.base_address) - self.assertEqual(self.session.find_base_address(m.name + "_does_not_exist$#@$"), 0) - - def test_memory_access(self): - result = {} - event = threading.Event() - def on_message(message, data): - self.assertEqual(message['type'], 'send') - result['address'] = int(message['payload'], 16) - event.set() - - script = self.session.create_script(name="test-memory-access", source="""\ -hello = Memory.allocUtf8String("Hello"); -send(hello); -""") - script.on('message', on_message) - script.load() - event.wait() - hello_address = result['address'] - - self.assertListEqual([x for x in iterbytes(self.session.read_bytes(hello_address, 6))], - [0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x00]) - self.assertEqual(self.session.read_utf8(hello_address), "Hello") - - self.session.write_bytes(hello_address, b"Yo\x00") - self.assertListEqual([x for x in iterbytes(self.session.read_bytes(hello_address, 6))], - [0x59, 0x6f, 0x00, 0x6c, 0x6f, 0x00]) - self.assertEqual(self.session.read_utf8(hello_address), "Yo") - self.session.write_utf8(hello_address, "Hei") - self.assertListEqual([x for x in iterbytes(self.session.read_bytes(hello_address, 6))], - [0x48, 0x65, 0x69, 0x00, 0x6f, 0x00]) - self.assertEqual(self.session.read_utf8(hello_address), "Hei") - - script.off('message', on_message) - script.unload() - - def test_enumerate_module_exports(self): - m = self.session.enumerate_modules()[1] - exports = m.enumerate_exports() - e = exports[0] - self.assertIsInstance(repr(e), str) - self.assertIsInstance(str(e), str) - - def test_enumerate_module_ranges(self): - m = self.session.enumerate_modules()[1] - ranges = m.enumerate_ranges('r--') - r = ranges[0] - self.assertIsInstance(repr(r), str) - self.assertIsInstance(str(r), str) - -if sys.version_info[0] >= 3: - iterbytes = lambda x: iter(x) -else: - def iterbytes(data): - return (ord(char) for char in data) - if __name__ == '__main__': unittest.main() diff --git a/tests/test_discoverer.py b/tests/test_discoverer.py deleted file mode 100644 index b147687..0000000 --- a/tests/test_discoverer.py +++ /dev/null @@ -1,56 +0,0 @@ -# -*- coding: utf-8 -*- - -from .data import target_program -import platform -import subprocess -import threading -try: - import unittest2 as unittest -except: - import unittest - -import frida -from frida.application import Reactor -from frida.discoverer import Discoverer, UI - - -class TestDiscoverer(unittest.TestCase): - @classmethod - def setUpClass(cls): - system = platform.system() - cls.target = subprocess.Popen([target_program], stdin=subprocess.PIPE) - cls.session = frida.attach(cls.target.pid) - - @classmethod - def tearDownClass(cls): - cls.session.detach() - cls.target.terminate() - cls.target.stdin.close() - - def test_basics(self): - test_ui = TestUI() - reactor = Reactor(lambda reactor: test_ui.on_result.wait()) - def start(): - d = Discoverer(reactor) - d.start(self.session, test_ui) - reactor.schedule(d.stop, 0.1) - reactor.schedule(start) - reactor.run() - self.assertIsInstance(test_ui.module_functions, dict) - self.assertIsInstance(test_ui.dynamic_functions, list) - -class TestUI(UI): - def __init__(self): - super(UI, self).__init__() - self.module_functions = None - self.dynamic_functions = None - self.on_result = threading.Event() - - def on_sample_result(self, module_functions, dynamic_functions): - self.module_functions = module_functions - self.dynamic_functions = dynamic_functions - self.on_result.set() - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/test_rpc.py b/tests/test_rpc.py index 56b52b3..fa6b1d0 100644 --- a/tests/test_rpc.py +++ b/tests/test_rpc.py @@ -1,6 +1,5 @@ # -*- coding: utf-8 -*- -from .data import target_program import platform import subprocess import sys @@ -13,8 +12,10 @@ except: import frida +from .data import target_program -class TestCore(unittest.TestCase): + +class TestRpc(unittest.TestCase): @classmethod def setUp(cls): system = platform.system() @@ -26,6 +27,7 @@ class TestCore(unittest.TestCase): cls.session.detach() cls.target.terminate() cls.target.stdin.close() + cls.target.wait() def test_basics(self): script = self.session.create_script(name="test-rpc", source="""\ @@ -116,6 +118,7 @@ rpc.exports = { m = self.assertRaisesRegex if sys.version_info[0] >= 3 else self.assertRaisesRegexp m(frida.InvalidOperationError, "script is destroyed", operation) + if sys.version_info[0] >= 3: iterbytes = lambda x: iter(x) else: diff --git a/tests/test_tracer.py b/tests/test_tracer.py deleted file mode 100644 index bb1669a..0000000 --- a/tests/test_tracer.py +++ /dev/null @@ -1,45 +0,0 @@ -# -*- coding: utf-8 -*- - -from .data import target_program -import platform -import subprocess -import threading -try: - import unittest2 as unittest -except: - import unittest - -import frida -from frida.application import Reactor -from frida.tracer import Tracer, TracerProfileBuilder, MemoryRepository, UI - - -class TestTracer(unittest.TestCase): - @classmethod - def setUpClass(cls): - system = platform.system() - cls.target = subprocess.Popen([target_program], stdin=subprocess.PIPE) - cls.session = frida.attach(cls.target.pid) - - @classmethod - def tearDownClass(cls): - cls.session.detach() - cls.target.terminate() - cls.target.stdin.close() - - def test_basics(self): - done = threading.Event() - reactor = Reactor(lambda reactor: done.wait()) - def start(): - tp = TracerProfileBuilder().include("open*") - t = Tracer(reactor, MemoryRepository(), tp.build()) - targets = t.start_trace(self.session, UI()) - t.stop() - reactor.stop() - done.set() - reactor.schedule(start) - reactor.run() - - -if __name__ == '__main__': - unittest.main()