frida/frida-python
1
0 Fork 0
mirror of https://github.com/frida/frida-python.git synced 2026-11-07 09:35:57 -10:00
Code Issues Packages Projects Releases Wiki Activity

Drop convenience APIs and split out tools (#142)

Browse Source 
Most use-cases will need to use some JS API that we're not bridging, so
it's better to keep our bindings simple.

It has also become clear that our CLI tools should be split out so Frida-based
tools don't have to depend on colorama, prompt_toolkit, and pygments.
This commit is contained in:
Ole André Vadla Ravnås authored and GitHub committed 2018-07-10 20:08:56 +02:00
1 parent 4fcb17c450
commit 72899a4315
36 files changed
+81 -3225
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitignore
+3 -1
View File
@@ -1,2 +1,4 @@
*.pyc
.idea/
/*.egg-info/
/build/
/dist/
README.md
+4 -4
View File
@@ -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
```
examples/bytecode.py
+2
View File
@@ -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';
examples/child_gating.py
+5 -2
View File
@@ -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()
examples/detached.py
+4 -1
View File
@@ -1,8 +1,11 @@
# -*- coding: utf-8 -*-
from __future__ import print_function
import frida
import sys
import frida
def on_detached():
print("on_detached")
examples/inject_library/inject_blob.py
+4 -1
View File
@@ -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)
examples/inject_library/inject_file.py
+4 -1
View File
@@ -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)
examples/rpc.py
+2
View File
@@ -1,7 +1,9 @@
# -*- coding: utf-8 -*-
from __future__ import print_function
import frida
session = frida.attach("Twitter")
script = session.create_script("""\
'use strict';
frida-python-common.props
+4 -11
View File
@@ -17,34 +17,27 @@ mkdir $(OutDir)frida || exit 1
copy /B /Y "$(ProjectDir)src\frida\*.py" $(OutDir)frida || exit 1
echo. &gt; $(IntDir)frida-package.stamp</Command>
<Outputs Condition="'$(Configuration)|$(Platform)'=='Debug|Win32'">$(IntDir)frida-package.stamp</Outputs>
<AdditionalInputs Condition="'$(Configuration)|$(Platform)'=='Debug|Win32'">$(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</AdditionalInputs>
<AdditionalInputs Condition="'$(Configuration)|$(Platform)'=='Debug|Win32'">$(ProjectDir)src\frida\core.py</AdditionalInputs>
<Command Condition="'$(Configuration)|$(Platform)'=='Release|Win32'">rmdir /S /Q $(OutDir)frida 2&gt;NUL
mkdir $(OutDir)frida || exit 1
copy /B /Y "$(ProjectDir)src\frida\*.py" $(OutDir)frida || exit 1
echo. &gt; $(IntDir)frida-package.stamp</Command>
<Outputs Condition="'$(Configuration)|$(Platform)'=='Release|Win32'">$(IntDir)frida-package.stamp</Outputs>
<AdditionalInputs Condition="'$(Configuration)|$(Platform)'=='Release|Win32'">$(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</AdditionalInputs>
<AdditionalInputs Condition="'$(Configuration)|$(Platform)'=='Release|Win32'">$(ProjectDir)src\frida\core.py</AdditionalInputs>
<Command Condition="'$(Configuration)|$(Platform)'=='Debug|x64'">rmdir /S /Q $(OutDir)frida 2&gt;NUL
mkdir $(OutDir)frida || exit 1
copy /B /Y "$(ProjectDir)src\frida\*.py" $(OutDir)frida || exit 1
echo. &gt; $(IntDir)frida-package.stamp</Command>
<Outputs Condition="'$(Configuration)|$(Platform)'=='Debug|x64'">$(IntDir)frida-package.stamp</Outputs>
<AdditionalInputs Condition="'$(Configuration)|$(Platform)'=='Debug|x64'">$(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</AdditionalInputs>
<AdditionalInputs Condition="'$(Configuration)|$(Platform)'=='Debug|x64'">$(ProjectDir)src\frida\core.py</AdditionalInputs>
<Command Condition="'$(Configuration)|$(Platform)'=='Release|x64'">rmdir /S /Q $(OutDir)frida 2&gt;NUL
mkdir $(OutDir)frida || exit 1
copy /B /Y "$(ProjectDir)src\frida\*.py" $(OutDir)frida || exit 1
echo. &gt; $(IntDir)frida-package.stamp</Command>
<Outputs Condition="'$(Configuration)|$(Platform)'=='Release|x64'">$(IntDir)frida-package.stamp</Outputs>
<AdditionalInputs Condition="'$(Configuration)|$(Platform)'=='Release|x64'">$(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</AdditionalInputs>
<AdditionalInputs Condition="'$(Configuration)|$(Platform)'=='Release|x64'">$(ProjectDir)src\frida\core.py</AdditionalInputs>
</CustomBuild>
<None Include="src\frida\core.py" />
<None Include="src\frida\application.py" />
<None Include="src\frida\lsd.py" />
<None Include="src\frida\ps.py" />
<None Include="src\frida\kill.py" />
<None Include="src\frida\discoverer.py" />
<None Include="src\frida\tracer.py" />
<None Include="src\frida\repl.py" />
</ItemGroup>
</Project>
frida-python27.vcxproj.filters
-21
View File
@@ -26,27 +26,6 @@
<None Include="src\frida\core.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\application.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\lsd.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\ps.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\kill.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\discoverer.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\tracer.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\repl.py">
<Filter>Source Files\frida</Filter>
</None>
</ItemGroup>
<ItemGroup>
<CustomBuild Include="src\frida\__init__.py">
frida-python36.vcxproj.filters
-21
View File
@@ -26,27 +26,6 @@
<None Include="src\frida\core.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\application.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\lsd.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\ps.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\kill.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\discoverer.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\tracer.py">
<Filter>Source Files\frida</Filter>
</None>
<None Include="src\frida\repl.py">
<Filter>Source Files\frida</Filter>
</None>
</ItemGroup>
<ItemGroup>
<CustomBuild Include="src\frida\__init__.py">
src/frida/__init__.py → frida/__init__.py
+18 -7
View File
@@ -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()
src/frida/core.py → frida/core.py
+11 -268
View File
@@ -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)
src/frida/meson.build → frida/meson.build
-7
View File
@@ -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'))
meson.build
+1
View File
@@ -75,3 +75,4 @@ configure_file(input: 'config.h.in',
configuration: cdata)
subdir('src')
subdir('frida')
src/setup.py → setup.py
+13 -23
View File
@@ -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
src/frida/application.py
-437
View File
@@ -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()
src/frida/discoverer.py
-203
View File
@@ -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()
src/frida/kill.py
-35
View File
@@ -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()
src/frida/lsd.py
-88
View File
@@ -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()
src/frida/ps.py
-125
View File
@@ -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()
src/frida/repl.py
-765
View File
@@ -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()
src/frida/tracer.py
-923
View File
@@ -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<module>[^\s!]+)!(?P<offset>(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()
src/meson.build
-1
View File
@@ -1,5 +1,4 @@
subdir('frida')
subdir('scripts')
extra_link_args = []
if host_os_family == 'darwin'
src/scripts/frida-discover.in
-11
View File
@@ -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()
src/scripts/frida-kill.in
-10
View File
@@ -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()
src/scripts/frida-ls-devices.in
-10
View File
@@ -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()
src/scripts/frida-ps.in
-10
View File
@@ -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()
src/scripts/frida-trace.in
-11
View File
@@ -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()
src/scripts/frida.in
-10
View File
@@ -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()
src/scripts/meson.build
-24
View File
@@ -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
tests/__init__.py
+1 -2
View File
@@ -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
tests/test_core.py
-89
View File
@@ -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()
tests/test_discoverer.py
-56
View File
@@ -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()
tests/test_rpc.py
+5 -2
View File
@@ -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:
tests/test_tracer.py
-45
View File
@@ -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()