admin/frida-frida-python
1
0
Fork 0
mirror of https://github.com/frida/frida-python synced 2026-06-08 14:16:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Rename Process -> Session and expose create_script()

Browse Source 
Also follow the same naming conventions as in frida-node.
This commit is contained in:
Ole André Vadla Ravnås
2015-03-19 16:40:31 +01:00
parent 617814de3f
commit ac7f3432e1
8 changed files with 91 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/frida/application.py
+9 -9
View File
@@ -49,8 +49,8 @@ class ConsoleApplication(object):
self._schedule_on_device_lost = lambda: self._reactor.schedule(self._on_device_lost)
self._spawned_pid = None
self._spawned_argv = None
self._process = None
self._schedule_on_process_detached = lambda: self._reactor.schedule(self._on_process_detached)
self._session = None
self._schedule_on_session_detached = lambda: self._reactor.schedule(self._on_session_detached)
self._started = False
self._resumed = False
self._reactor = Reactor(run_until_return)
@@ -86,10 +86,10 @@ class ConsoleApplication(object):
self._reactor.run()
if self._started:
self._stop()
if self._process is not None:
self._process.off('detached', self._schedule_on_process_detached)
self._process.detach()
self._process = None
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:
self._device.kill(self._spawned_pid)
if self._device is not None:
@@ -143,8 +143,8 @@ class ConsoleApplication(object):
else:
attach_target = target_value
self._update_status("Attaching...")
self._process = self._device.attach(attach_target)
self._process.on('detached', self._schedule_on_process_detached)
self._session = self._device.attach(attach_target)
self._session.on('detached', self._schedule_on_session_detached)
except Exception as e:
self._update_status("Failed to attach: %s" % e)
self._exit(1)
@@ -162,7 +162,7 @@ class ConsoleApplication(object):
print("Device disconnected.")
self._exit(1)
def _on_process_detached(self):
def _on_session_detached(self):
print("Target process terminated.")
self._exit(1)
src/frida/core.py
+43 -34
View File
@@ -8,17 +8,17 @@ import threading
class DeviceManager(object):
def __init__(self, manager):
self._manager = manager
def __init__(self, impl):
self._impl = impl
def __repr__(self):
return repr(self._manager)
return repr(self._impl)
def enumerate_devices(self):
return [Device(device) for device in self._manager.enumerate_devices()]
return [Device(device) for device in self._impl.enumerate_devices()]
def get_device(self, device_id):
devices = self._manager.enumerate_devices()
devices = self._impl.enumerate_devices()
if device_id is None:
return Device(devices[0])
for device in devices:
@@ -27,10 +27,10 @@ class DeviceManager(object):
raise ValueError("device not found")
def on(self, signal, callback):
self._manager.on(signal, callback)
self._impl.on(signal, callback)
def off(self, signal, callback):
self._manager.off(signal, callback)
self._impl.off(signal, callback)
class Device(object):
def __init__(self, device):
@@ -38,17 +38,17 @@ class Device(object):
self.name = device.name
self.icon = device.icon
self.type = device.type
self._device = device
self._impl = device
def __repr__(self):
return repr(self._device)
return repr(self._impl)
def enumerate_processes(self):
return self._device.enumerate_processes()
return self._impl.enumerate_processes()
def get_process(self, process_name):
process_name_lc = process_name.lower()
matching = [process for process in self._device.enumerate_processes() if fnmatch.fnmatchcase(process.name.lower(), process_name_lc)]
matching = [process for process in self._impl.enumerate_processes() if fnmatch.fnmatchcase(process.name.lower(), process_name_lc)]
if len(matching) == 1:
return matching[0]
elif len(matching) > 1:
@@ -57,22 +57,22 @@ class Device(object):
raise ValueError("process not found")
def spawn(self, command_line):
return self._device.spawn(command_line)
return self._impl.spawn(command_line)
def resume(self, target):
self._device.resume(self._pid_of(target))
self._impl.resume(self._pid_of(target))
def kill(self, target):
self._device.kill(self._pid_of(target))
self._impl.kill(self._pid_of(target))
def attach(self, target):
return Process(self._device.attach(self._pid_of(target)))
return Session(self._impl.attach(self._pid_of(target)))
def on(self, signal, callback):
self._device.on(signal, callback)
self._impl.on(signal, callback)
def off(self, signal, callback):
self._device.off(signal, callback)
self._impl.off(signal, callback)
def _pid_of(self, target):
if isinstance(target, numbers.Number):
@@ -96,19 +96,19 @@ class FunctionContainer(object):
def _do_ensure_function(self, address):
raise NotImplementedError("not implemented")
class Process(FunctionContainer):
def __init__(self, session):
super(Process, self).__init__()
self.session = session
class Session(FunctionContainer):
def __init__(self, impl):
super(Session, self).__init__()
self._impl = impl
self._modules = None
self._module_map = None
def detach(self):
self.session.detach()
self._impl.detach()
def enumerate_modules(self):
if self._modules is None:
script = self.session.create_script(
script = self.create_script(
"""
var modules = [];
Process.enumerateModules({
@@ -120,14 +120,14 @@ class Process(FunctionContainer):
}
});
""")
self._modules = [Module(data['name'], int(data['base'], 16), data['size'], data['path'], self.session) for data in _execute_script(script)]
self._modules = [Module(data['name'], int(data['base'], 16), data['size'], data['path'], self) for data in _execute_script(script)]
return self._modules
"""
@param protection example '--x'
"""
def enumerate_ranges(self, protection):
script = self.session.create_script(
script = self.create_script(
"""
var ranges = [];
Process.enumerateRanges(\"%s\", {
@@ -141,19 +141,12 @@ Process.enumerateRanges(\"%s\", {
""" % protection)
return [Range(int(data['base'], 16), data['size'], data['protection']) for data in _execute_script(script)]
def _exec_script(self, script_source, post_hook = None):
script = self.session.create_script(script_source)
return _execute_script(script, post_hook)
def find_base_address(self, module_name):
return int(self._exec_script("var p = Module.findBaseAddress(\"%s\"); send(p !== null ? p.toString() : \"0\");" % module_name), 16)
def read_bytes(self, address, length):
return self._exec_script("send(null, Memory.readByteArray(ptr(\"%u\"), %u));" % (address, length))
def read_utf8(self, address, length = -1):
return self._exec_script("send(Memory.readUtf8String(ptr(\"%u\"), %u));" % (address, length))
def write_bytes(self, address, data):
script = \
"""
@@ -170,6 +163,9 @@ recv(function (data) {
self._exec_script(script, send_data)
def read_utf8(self, address, length = -1):
return self._exec_script("send(Memory.readUtf8String(ptr(\"%u\"), %u));" % (address, length))
def write_utf8(self, address, string):
script = \
"""
@@ -184,11 +180,18 @@ recv(function (string) {
self._exec_script(script, send_data)
def create_script(self, source):
return self._impl.create_script(source)
def on(self, signal, callback):
self.session.on(signal, callback)
self._impl.on(signal, callback)
def off(self, signal, callback):
self.session.off(signal, callback)
self._impl.off(signal, callback)
def _exec_script(self, script_source, post_hook = None):
script = self.create_script(script_source)
return _execute_script(script, post_hook)
def _do_ensure_function(self, absolute_address):
if self._module_map is None:
@@ -201,6 +204,12 @@ recv(function (string) {
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 super(Session, self).__getattr__(self, attr)
class Module(FunctionContainer):
def __init__(self, name, base_address, size, path, session):
super(Module, self).__init__()
src/frida/discoverer.py
+6 -6
View File
@@ -18,11 +18,11 @@ class Discoverer(object):
pass
self._script = None
def start(self, process, ui):
def start(self, session, ui):
def on_message(message, data):
self._reactor.schedule(lambda: self._process_message(message, data, process, ui))
self._reactor.schedule(lambda: self._process_message(message, data, session, ui))
source = self._create_discover_script()
self._script = process.session.create_script(source)
self._script = session.create_script(source)
self._script.on('message', on_message)
self._script.load()
@@ -104,7 +104,7 @@ sampler = new Sampler();
setTimeout(function () { sampler.start(); }, 0);
"""
def _process_message(self, message, data, process, ui):
def _process_message(self, message, data, session, ui):
if message['type'] == 'send':
stanza = message['payload']
name = stanza['name']
@@ -117,7 +117,7 @@ setTimeout(function () { sampler.start(); }, 0);
dynamic_functions = []
for address, count in payload['result'].items():
address = int(address, 16)
function = process.ensure_function(address)
function = session.ensure_function(address)
if isinstance(function, ModuleFunction):
functions = module_functions.get(function.module, [])
if len(functions) == 0:
@@ -166,7 +166,7 @@ def main():
def _start(self):
self._update_status("Injecting script...")
self._discoverer = Discoverer(self._reactor)
self._discoverer.start(self._process, self)
self._discoverer.start(self._session, self)
def _stop(self):
print("Stopping...")
src/frida/repl.py
+1 -1
View File
@@ -26,7 +26,7 @@ def main():
def _start(self):
def on_message(message, data):
self._reactor.schedule(lambda: self._process_message(message, data))
self._script = self._process.session.create_script(self._create_repl_script())
self._script = self._session.create_script(self._create_repl_script())
self._script.on('message', on_message)
self._script.load()
if self._spawned_argv is not None:
src/frida/tracer.py
+8 -8
View File
@@ -52,8 +52,8 @@ class TracerProfile(object):
def __init__(self, spec):
self._spec = spec
def resolve(self, process):
all_modules = process.enumerate_modules()
def resolve(self, session):
all_modules = session.enumerate_modules()
working_set = set()
for (operation, scope, param) in self._spec:
if scope == 'module':
@@ -68,8 +68,8 @@ class TracerProfile(object):
working_set = self._exclude_function(param, working_set)
elif scope == 'rel_address':
if operation == 'include':
abs_address = process.find_base_address(param['module']) + param['offset']
working_set.add(process.ensure_function(abs_address))
abs_address = session.find_base_address(param['module']) + param['offset']
working_set.add(session.ensure_function(abs_address))
return list(working_set)
def _include_module(self, glob, all_modules):
@@ -109,7 +109,7 @@ class Tracer(object):
self._profile = profile
self._script = None
def start_trace(self, process, ui):
def start_trace(self, session, ui):
def on_create(*args):
ui.on_trace_handler_create(*args)
self._repository.on_create(on_create)
@@ -135,10 +135,10 @@ class Tracer(object):
self._reactor.schedule(lambda: self._process_message(message, data, ui))
ui.on_trace_progress('resolve')
working_set = self._profile.resolve(process)
working_set = self._profile.resolve(session)
source = self._create_trace_script()
ui.on_trace_progress('instrument')
self._script = process.session.create_script(source)
self._script = session.create_script(source)
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)]:
@@ -506,7 +506,7 @@ def main():
def _start(self):
self._tracer = Tracer(self._reactor, FileRepository(), self._profile)
self._targets = self._tracer.start_trace(self._process, self)
self._targets = self._tracer.start_trace(self._session, self)
def _stop(self):
print("Stopping...")
tests/test_core.py
+18 -18
View File
@@ -20,11 +20,11 @@ class TestCore(unittest.TestCase):
cls.target = subprocess.Popen([r"C:\Windows\notepad.exe"])
else:
cls.target = subprocess.Popen(["/bin/cat"])
cls.process = frida.attach(cls.target.pid)
cls.session = frida.attach(cls.target.pid)
@classmethod
def tearDownClass(cls):
cls.process.detach()
cls.session.detach()
cls.target.terminate()
def test_enumerate_devices(self):
@@ -32,23 +32,23 @@ class TestCore(unittest.TestCase):
self.assertTrue(len(devices) > 0)
def test_enumerate_modules(self):
modules = self.process.enumerate_modules()
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.process.enumerate_ranges('r--')
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.process.enumerate_modules()[0]
self.assertEqual(self.process.find_base_address(m.name), m.base_address)
self.assertEqual(self.process.find_base_address(m.name + "_does_not_exist$#@$"), 0)
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 = {}
@@ -58,7 +58,7 @@ class TestCore(unittest.TestCase):
result['address'] = int(message['payload'], 16)
event.set()
script = self.process.session.create_script("""\
script = self.session.create_script("""\
hello = Memory.allocUtf8String("Hello");
send(hello);
""")
@@ -67,31 +67,31 @@ send(hello);
event.wait()
hello_address = result['address']
self.assertListEqual([x for x in iterbytes(self.process.read_bytes(hello_address, 6))],
self.assertListEqual([x for x in iterbytes(self.session.read_bytes(hello_address, 6))],
[0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x00])
self.assertEqual(self.process.read_utf8(hello_address), "Hello")
self.assertEqual(self.session.read_utf8(hello_address), "Hello")
self.process.write_bytes(hello_address, b"Yo\x00")
self.assertListEqual([x for x in iterbytes(self.process.read_bytes(hello_address, 6))],
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.process.read_utf8(hello_address), "Yo")
self.process.write_utf8(hello_address, "Hei")
self.assertListEqual([x for x in iterbytes(self.process.read_bytes(hello_address, 6))],
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.process.read_utf8(hello_address), "Hei")
self.assertEqual(self.session.read_utf8(hello_address), "Hei")
script.off('message', on_message)
script.unload()
def test_enumerate_module_exports(self):
m = self.process.enumerate_modules()[1]
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.process.enumerate_modules()[1]
m = self.session.enumerate_modules()[1]
ranges = m.enumerate_ranges('r--')
r = ranges[0]
self.assertIsInstance(repr(r), str)
tests/test_discoverer.py
+3 -3
View File
@@ -21,11 +21,11 @@ class TestDiscoverer(unittest.TestCase):
cls.target = subprocess.Popen([r"C:\Windows\notepad.exe"])
else:
cls.target = subprocess.Popen(["/bin/cat"])
cls.process = frida.attach(cls.target.pid)
cls.session = frida.attach(cls.target.pid)
@classmethod
def tearDownClass(cls):
cls.process.detach()
cls.session.detach()
cls.target.terminate()
def test_basics(self):
@@ -33,7 +33,7 @@ class TestDiscoverer(unittest.TestCase):
reactor = Reactor(test_ui.on_result.wait)
def start():
d = Discoverer(reactor)
d.start(self.process, test_ui)
d.start(self.session, test_ui)
reactor.schedule(d.stop, 0.1)
reactor.schedule(start)
reactor.run()
tests/test_tracer.py
+3 -3
View File
@@ -21,11 +21,11 @@ class TestTracer(unittest.TestCase):
cls.target = subprocess.Popen([r"C:\Windows\notepad.exe"])
else:
cls.target = subprocess.Popen(["/bin/cat"])
cls.process = frida.attach(cls.target.pid)
cls.session = frida.attach(cls.target.pid)
@classmethod
def tearDownClass(cls):
cls.process.detach()
cls.session.detach()
cls.target.terminate()
def test_basics(self):
@@ -34,7 +34,7 @@ class TestTracer(unittest.TestCase):
def start():
tp = TracerProfileBuilder().include("open*")
t = Tracer(reactor, MemoryRepository(), tp.build())
targets = t.start_trace(self.process, UI())
targets = t.start_trace(self.session, UI())
t.stop()
reactor.stop()
reactor.schedule(start)