From fc00ac2b753d5d506ac230710b08322ca703558e Mon Sep 17 00:00:00 2001 From: Yotam Nachum Date: Sat, 29 Jan 2022 00:16:53 +0200 Subject: [PATCH] Automatically format code with Black and Isort --- examples/bytecode.py | 8 +- examples/channels.py | 1 - examples/child_gating.py | 11 +- examples/cpushark/AppDelegate.py | 1 + examples/cpushark/Capture.py | 86 ++++--- examples/cpushark/CpuShark.py | 2 + examples/cpushark/MainWindowController.py | 7 +- examples/cpushark/ProcessList.py | 4 +- examples/cpushark/setup.py | 16 +- examples/crash_reporting.py | 6 +- examples/detached.py | 9 +- examples/enumerate_applications.py | 15 +- examples/enumerate_processes.py | 17 +- examples/get_frontmost_application.py | 13 +- examples/inject_library/inject_blob.py | 1 + examples/inject_library/inject_file.py | 1 + examples/portal_client.py | 68 +++--- examples/portal_server.py | 120 +++++----- examples/query_system_parameters.py | 8 +- examples/rpc.py | 7 +- examples/session_persist_timeout.py | 13 +- examples/snapshot.py | 3 +- frida/__init__.py | 7 +- frida/core.py | 93 ++++---- pyproject.toml | 6 + setup.py | 122 +++++----- tests/data/__init__.py | 11 +- tests/test_core.py | 16 +- tests/test_pep503_page_parser.py | 273 +++++++++++----------- tests/test_rpc.py | 42 ++-- 30 files changed, 522 insertions(+), 465 deletions(-) create mode 100644 pyproject.toml diff --git a/examples/bytecode.py b/examples/bytecode.py index 0f820b9..d7abdfa 100644 --- a/examples/bytecode.py +++ b/examples/bytecode.py @@ -1,14 +1,16 @@ import frida - system_session = frida.attach(0) -bytecode = system_session.compile_script(name="bytecode-example", source="""\ +bytecode = system_session.compile_script( + name="bytecode-example", + source="""\ rpc.exports = { listThreads: function () { return Process.enumerateThreadsSync(); } }; -""") +""", +) session = frida.attach("Twitter") script = session.create_script_from_bytes(bytecode) diff --git a/examples/channels.py b/examples/channels.py index 16a1d88..fe8dbb4 100644 --- a/examples/channels.py +++ b/examples/channels.py @@ -1,6 +1,5 @@ import frida - device = frida.get_usb_device() channel = device.open_channel("tcp:21") diff --git a/examples/child_gating.py b/examples/child_gating.py index 2006851..0e90ac5 100644 --- a/examples/child_gating.py +++ b/examples/child_gating.py @@ -1,8 +1,9 @@ import threading -import frida from frida_tools.application import Reactor +import frida + class Application: def __init__(self): @@ -27,7 +28,7 @@ class Application: "SNAKE": "mushroom-mushroom", } print(f"✔ spawn(argv={argv})") - pid = self._device.spawn(argv, env=env, stdio='pipe') + pid = self._device.spawn(argv, env=env, stdio="pipe") self._instrument(pid) def _stop_if_idle(self): @@ -41,7 +42,8 @@ class Application: print("✔ enable_child_gating()") session.enable_child_gating() print("✔ create_script()") - script = session.create_script("""\ + script = session.create_script( + """\ Interceptor.attach(Module.getExportByName(null, 'open'), { onEnter: function (args) { send({ @@ -50,7 +52,8 @@ Interceptor.attach(Module.getExportByName(null, 'open'), { }); } }); -""") +""" + ) script.on("message", lambda message, data: self._reactor.schedule(lambda: self._on_message(pid, message))) print("✔ load()") script.load() diff --git a/examples/cpushark/AppDelegate.py b/examples/cpushark/AppDelegate.py index 8029639..b1d7482 100644 --- a/examples/cpushark/AppDelegate.py +++ b/examples/cpushark/AppDelegate.py @@ -2,6 +2,7 @@ from Cocoa import NSApp from Foundation import NSObject from MainWindowController import MainWindowController + class AppDelegate(NSObject): def applicationDidFinishLaunching_(self, notification): window = MainWindowController() diff --git a/examples/cpushark/Capture.py b/examples/cpushark/Capture.py index 1eee57a..16e27a4 100644 --- a/examples/cpushark/Capture.py +++ b/examples/cpushark/Capture.py @@ -1,11 +1,13 @@ import bisect -from Foundation import NSAutoreleasePool, NSObject, NSThread -from PyObjCTools import AppHelper import re import struct +from Foundation import NSAutoreleasePool, NSObject, NSThread +from PyObjCTools import AppHelper + PROBE_CALLS = re.compile(r"^\/stalker\/probes\/(.*?)\/calls$") + class Capture(NSObject): def __new__(cls, device): return cls.alloc().initWithDevice_(device) @@ -31,7 +33,7 @@ class Capture(NSObject): def attachToProcess_triggerPort_(self, process, triggerPort): assert self.state == CaptureState.DETACHED self._updateState_(CaptureState.ATTACHING) - NSThread.detachNewThreadSelector_toTarget_withObject_('_doAttachWithParams:', self, (process.pid, triggerPort)) + NSThread.detachNewThreadSelector_toTarget_withObject_("_doAttachWithParams:", self, (process.pid, triggerPort)) def detach(self): assert self.state == CaptureState.ATTACHED @@ -40,10 +42,10 @@ class Capture(NSObject): self.session = None self.script = None self._updateState_(CaptureState.DETACHED) - NSThread.detachNewThreadSelector_toTarget_withObject_('_doDetachWithParams:', self, (session, script)) + NSThread.detachNewThreadSelector_toTarget_withObject_("_doDetachWithParams:", self, (session, script)) def _post(self, message): - NSThread.detachNewThreadSelector_toTarget_withObject_('_doPostWithParams:', self, (self.script, message)) + NSThread.detachNewThreadSelector_toTarget_withObject_("_doPostWithParams:", self, (self.script, message)) def _updateState_(self, newState): self.state = newState @@ -57,11 +59,9 @@ class Capture(NSObject): error = None try: session = self.device.attach(pid) - session.on('detached', self._onSessionDetached) - script = session.create_script(name="cpushark", source=SCRIPT_TEMPLATE % { - 'trigger_port': triggerPort - }) - script.on('message', self._onScriptMessage) + session.on("detached", self._onSessionDetached) + script = session.create_script(name="cpushark", source=SCRIPT_TEMPLATE % {"trigger_port": triggerPort}) + script.on("message", self._onScriptMessage) script.load() except Exception as e: if session is not None: @@ -113,15 +113,15 @@ class Capture(NSObject): self._updateState_(CaptureState.DETACHED) def _sessionDidReceiveMessage_data_(self, message, data): - if message['type'] == 'send': - stanza = message['payload'] - fromAddress = stanza['from'] - name = stanza['name'] - if fromAddress == "/process/modules" and name == '+sync': - self.modules._sync(stanza['payload']) - elif fromAddress == "/stalker/calls" and name == '+add': - self.calls._add_(stanza['payload']) - elif fromAddress == "/interceptor/functions" and name == '+add': + if message["type"] == "send": + stanza = message["payload"] + fromAddress = stanza["from"] + name = stanza["name"] + if fromAddress == "/process/modules" and name == "+sync": + self.modules._sync(stanza["payload"]) + elif fromAddress == "/stalker/calls" and name == "+add": + self.calls._add_(stanza["payload"]) + elif fromAddress == "/interceptor/functions" and name == "+add": self.recvTotal += 1 self._delegate.captureRecvTotalDidChange() else: @@ -136,11 +136,13 @@ class Capture(NSObject): def _onScriptMessage(self, message, data): AppHelper.callAfter(self._sessionDidReceiveMessage_data_, message, data) + class CaptureState: DETACHED = 1 ATTACHING = 2 ATTACHED = 3 + class Modules: def __init__(self): self._modules = [] @@ -148,11 +150,11 @@ class Modules: def _sync(self, payload): modules = [] - for item in payload['items']: - modules.append(Module(item['name'], int(item['base'], 16), item['size'])) + for item in payload["items"]: + modules.append(Module(item["name"], int(item["base"], 16), item["size"])) modules.sort(lambda x, y: x.address - y.address) self._modules = modules - self._indices = [ m.address for m in modules ] + self._indices = [m.address for m in modules] def lookup(self, addr): idx = bisect.bisect(self._indices, addr) @@ -163,6 +165,7 @@ class Modules: return None return m + class Module: def __init__(self, name, address, size): self.name = name @@ -172,6 +175,7 @@ class Module: def __repr__(self): return "(%d, %d, %s)" % (self.address, self.size, self.name) + class Calls(NSObject): def __new__(cls, capture): return cls.alloc().initWithCapture_(capture) @@ -192,28 +196,16 @@ class Calls(NSObject): self._delegate = delegate def addProbe_(self, func): - self.capture._post({ - 'to': "/stalker/probes", - 'name': '+add', - 'payload': { - 'address': "0x%x" % func.address - } - }) + self.capture._post({"to": "/stalker/probes", "name": "+add", "payload": {"address": "0x%x" % func.address}}) self._probes[func.address] = func def removeProbe_(self, func): - self.capture._post({ - 'to': "/stalker/probes", - 'name': '+remove', - 'payload': { - 'address': "0x%x" % func.address - } - }) + self.capture._post({"to": "/stalker/probes", "name": "+remove", "payload": {"address": "0x%x" % func.address}}) self._probes.pop(func.address, None) def _add_(self, data): modules = self.capture.modules - for rawTarget, count in data['summary'].items(): + for rawTarget, count in data["summary"].items(): target = int(rawTarget, 16) tm = self.getTargetModuleByModule_(modules.lookup(target)) if tm is not None: @@ -237,13 +229,13 @@ class Calls(NSObject): return x.total - y.total def _handleStanza_(self, stanza): - m = PROBE_CALLS.match(stanza['from']) + m = PROBE_CALLS.match(stanza["from"]) if m is not None: func = self._probes.get(int(m.groups()[0], 16), None) if func is not None: if len(func.calls) == 3: func.calls.pop(0) - func.calls.append(FunctionCall(func, stanza['payload']['args'])) + func.calls.append(FunctionCall(func, stanza["payload"]["args"])) self._delegate.callItemDidChange_(func) return True return False @@ -291,27 +283,28 @@ class Calls(NSObject): def outlineView_objectValueForTableColumn_byItem_(self, outlineView, tableColumn, item): identifier = tableColumn.identifier() if isinstance(item, TargetModule): - if identifier == 'name': + if identifier == "name": return item.module.name - elif identifier == 'total': + elif identifier == "total": return item.total else: return False elif isinstance(item, TargetFunction): - if identifier == 'name': + if identifier == "name": return item.name - elif identifier == 'total': + elif identifier == "total": return item.total else: return item.hasProbe else: - if identifier == 'name': + if identifier == "name": return item.summary - elif identifier == 'total': + elif identifier == "total": return "" else: return False + class TargetModule(NSObject): def __new__(cls, module): return cls.alloc().initWithModule_(module) @@ -332,6 +325,7 @@ class TargetModule(NSObject): self._functionByAddress[address] = f return f + class TargetFunction(NSObject): def __new__(cls, module, offset): return cls.alloc().initWithModule_offset_(module, offset) @@ -347,6 +341,7 @@ class TargetFunction(NSObject): self.calls = [] return self + class FunctionCall(NSObject): def __new__(cls, func, args): return cls.alloc().initWithFunction_args_(func, args) @@ -358,6 +353,7 @@ class FunctionCall(NSObject): self.summary = f"{func.name}({', '.join(args)})" return self + SCRIPT_TEMPLATE = """ var probes = Object.create(null); diff --git a/examples/cpushark/CpuShark.py b/examples/cpushark/CpuShark.py index dc898f2..a88780c 100644 --- a/examples/cpushark/CpuShark.py +++ b/examples/cpushark/CpuShark.py @@ -1,4 +1,5 @@ import sys + sys.path.insert(0, "/Users/oleavr/src/frida/build/frida-macos-universal/lib/python2.7/site-packages") import AppDelegate import Capture @@ -7,4 +8,5 @@ import ProcessList if __name__ == "__main__": from PyObjCTools import AppHelper + AppHelper.runEventLoop() diff --git a/examples/cpushark/MainWindowController.py b/examples/cpushark/MainWindowController.py index 63e14df..58328ae 100644 --- a/examples/cpushark/MainWindowController.py +++ b/examples/cpushark/MainWindowController.py @@ -1,8 +1,10 @@ -import frida from Capture import Capture, CaptureState, TargetFunction from Cocoa import NSRunCriticalAlertPanel, NSUserDefaults, NSWindowController, objc from ProcessList import ProcessList +import frida + + class MainWindowController(NSWindowController): processCombo = objc.IBOutlet() triggerField = objc.IBOutlet() @@ -26,7 +28,7 @@ class MainWindowController(NSWindowController): def windowDidLoad(self): NSWindowController.windowDidLoad(self) - device = [device for device in frida.get_device_manager().enumerate_devices() if device.type == 'local'][0] + device = [device for device in frida.get_device_manager().enumerate_devices() if device.type == "local"][0] self.processList = ProcessList(device) self.capture = Capture(device) self.processCombo.setUsesDataSource_(True) @@ -127,4 +129,3 @@ class MainWindowController(NSWindowController): def callItemDidChange_(self, item): self.callTableView.reloadItem_reloadChildren_(item, True) - diff --git a/examples/cpushark/ProcessList.py b/examples/cpushark/ProcessList.py index cec6567..a846d3b 100644 --- a/examples/cpushark/ProcessList.py +++ b/examples/cpushark/ProcessList.py @@ -1,4 +1,5 @@ -from Foundation import NSObject, NSNotFound +from Foundation import NSNotFound, NSObject + class ProcessList(NSObject): def __new__(cls, device): @@ -30,4 +31,3 @@ class ProcessList(NSObject): def comboBox_indexOfItemWithStringValue_(self, comboBox, value): return self._processIndexByName.get(value.lower(), NSNotFound) - diff --git a/examples/cpushark/setup.py b/examples/cpushark/setup.py index dc3bce5..9506af8 100644 --- a/examples/cpushark/setup.py +++ b/examples/cpushark/setup.py @@ -5,21 +5,21 @@ Usage: python setup.py py2app """ -from setuptools import setup import py2app +from setuptools import setup plist = dict( - CFBundleShortVersionString='CpuShark v1', - CFBundleIconFile='CpuShark.icns', - CFBundleGetInfoString='CpuShark v1', - CFBundleIdentifier='com.tillitech.CpuShark', + CFBundleShortVersionString="CpuShark v1", + CFBundleIconFile="CpuShark.icns", + CFBundleGetInfoString="CpuShark v1", + CFBundleIdentifier="com.tillitech.CpuShark", CFBundleDocumentTypes=[], - CFBundleName='CpuShark' + CFBundleName="CpuShark", ) setup( app=["CpuShark.py"], data_files=["MainMenu.xib", "MainWindow.xib"], - options={'py2app': {'plist': plist}}, - setup_requires=['py2app'], + options={"py2app": {"plist": plist}}, + setup_requires=["py2app"], ) diff --git a/examples/crash_reporting.py b/examples/crash_reporting.py index 3a387ee..736f672 100644 --- a/examples/crash_reporting.py +++ b/examples/crash_reporting.py @@ -7,14 +7,16 @@ def on_process_crashed(crash): print("on_process_crashed") print("\tcrash:", crash) + def on_detached(reason, crash): print("on_detached()") print("\treason:", reason) print("\tcrash:", crash) + device = frida.get_usb_device() -device.on('process-crashed', on_process_crashed) +device.on("process-crashed", on_process_crashed) session = device.attach("Hello") -session.on('detached', on_detached) +session.on("detached", on_detached) print("[*] Ready") sys.stdin.read() diff --git a/examples/detached.py b/examples/detached.py index 9b72709..638c2a8 100644 --- a/examples/detached.py +++ b/examples/detached.py @@ -6,15 +6,18 @@ import frida def on_detached(): print("on_detached") + def on_detached_with_reason(reason): print("on_detached_with_reason:", reason) + def on_detached_with_varargs(*args): print("on_detached_with_varargs:", args) + session = frida.attach("Twitter") print("attached") -session.on('detached', on_detached) -session.on('detached', on_detached_with_reason) -session.on('detached', on_detached_with_varargs) +session.on("detached", on_detached) +session.on("detached", on_detached_with_reason) +session.on("detached", on_detached_with_varargs) sys.stdin.read() diff --git a/examples/enumerate_applications.py b/examples/enumerate_applications.py index 59480ae..075df30 100644 --- a/examples/enumerate_applications.py +++ b/examples/enumerate_applications.py @@ -1,21 +1,24 @@ -import frida from pprint import pformat + from pygments import highlight from pygments.formatters import Terminal256Formatter from pygments.lexers import PythonLexer +import frida device = frida.get_usb_device() + def trim_icon(icon): result = dict(icon) - result['image'] = result['image'][0:16] + b"..." + result["image"] = result["image"][0:16] + b"..." return result -apps = device.enumerate_applications(scope='full') + +apps = device.enumerate_applications(scope="full") for app in apps: params = dict(app.parameters) - if 'icons' in params: - params['icons'] = [trim_icon(icon) for icon in params['icons']] + if "icons" in params: + params["icons"] = [trim_icon(icon) for icon in params["icons"]] parameters = highlight(pformat(params), PythonLexer(), Terminal256Formatter()).rstrip() - print(f"Application(identifier=\"{app.identifier}\", name=\"{app.name}\", pid={app.pid}, parameters={parameters})") + print(f'Application(identifier="{app.identifier}", name="{app.name}", pid={app.pid}, parameters={parameters})') diff --git a/examples/enumerate_processes.py b/examples/enumerate_processes.py index 588f35a..a78d8e0 100644 --- a/examples/enumerate_processes.py +++ b/examples/enumerate_processes.py @@ -1,20 +1,25 @@ -import frida from pprint import pformat + from pygments import highlight from pygments.formatters import Terminal256Formatter from pygments.lexers import PythonLexer +import frida device = frida.get_usb_device() + def trim_icon(icon): result = dict(icon) - result['image'] = result['image'][0:16] + b"..." + result["image"] = result["image"][0:16] + b"..." return result -processes = device.enumerate_processes(scope='full') + +processes = device.enumerate_processes(scope="full") for proc in processes: params = dict(proc.parameters) - if 'icons' in params: - params['icons'] = [trim_icon(icon) for icon in params['icons']] - print(f"Process(pid={proc.pid}, name=\"{proc.name}\", parameters={highlight(pformat(params), PythonLexer(), Terminal256Formatter()).rstrip()})") + if "icons" in params: + params["icons"] = [trim_icon(icon) for icon in params["icons"]] + print( + f'Process(pid={proc.pid}, name="{proc.name}", parameters={highlight(pformat(params), PythonLexer(), Terminal256Formatter()).rstrip()})' + ) diff --git a/examples/get_frontmost_application.py b/examples/get_frontmost_application.py index 04f512f..708fb11 100644 --- a/examples/get_frontmost_application.py +++ b/examples/get_frontmost_application.py @@ -1,22 +1,25 @@ -import frida from pprint import pformat + from pygments import highlight from pygments.formatters import Terminal256Formatter from pygments.lexers import PythonLexer +import frida device = frida.get_usb_device() + def trim_icon(icon): result = dict(icon) - result['image'] = result['image'][0:16] + b"..." + result["image"] = result["image"][0:16] + b"..." return result -app = device.get_frontmost_application(scope='full') + +app = device.get_frontmost_application(scope="full") if app is not None: params = dict(app.parameters) - if 'icons' in params: - params['icons'] = [trim_icon(icon) for icon in params['icons']] + if "icons" in params: + params["icons"] = [trim_icon(icon) for icon in params["icons"]] print(f"{app.identifier}:", highlight(pformat(params), PythonLexer(), Terminal256Formatter())) else: print("No frontmost application") diff --git a/examples/inject_library/inject_blob.py b/examples/inject_library/inject_blob.py index 0169c46..5d3c817 100644 --- a/examples/inject_library/inject_blob.py +++ b/examples/inject_library/inject_blob.py @@ -14,6 +14,7 @@ import frida def on_uninjected(id): print("on_uninjected id=%u" % id) + (target, library_path) = sys.argv[1:] device = frida.get_local_device() diff --git a/examples/inject_library/inject_file.py b/examples/inject_library/inject_file.py index bd72481..cd104ed 100644 --- a/examples/inject_library/inject_file.py +++ b/examples/inject_library/inject_file.py @@ -14,6 +14,7 @@ import frida def on_uninjected(id): print("on_uninjected id=%u" % id) + (target, library_path) = sys.argv[1:] device = frida.get_local_device() diff --git a/examples/portal_client.py b/examples/portal_client.py index 1dbbc21..38f524d 100644 --- a/examples/portal_client.py +++ b/examples/portal_client.py @@ -1,22 +1,20 @@ -import frida -from frida_tools.application import Reactor import json import sys +from frida_tools.application import Reactor + +import frida + class Application: def __init__(self, nick): self._reactor = Reactor(run_until_return=self._process_input) - token = { - 'nick': nick, - 'secret': "knock-knock" - } - self._device = frida.get_device_manager().add_remote_device("::1", - token=json.dumps(token)) + token = {"nick": nick, "secret": "knock-knock"} + self._device = frida.get_device_manager().add_remote_device("::1", token=json.dumps(token)) self._bus = self._device.bus - self._bus.on('message', lambda *args: self._reactor.schedule(lambda: self._on_bus_message(*args))) + self._bus.on("message", lambda *args: self._reactor.schedule(lambda: self._on_bus_message(*args))) self._channel = None self._prompt = "> " @@ -45,53 +43,43 @@ class Application: if text.startswith("/join "): if self._channel is not None: - self._bus.post({ - 'type': 'part', - 'channel': self._channel - }) + self._bus.post({"type": "part", "channel": self._channel}) channel = text[6:] self._channel = channel self._prompt = f"{channel} > " - self._bus.post({ - 'type': 'join', - 'channel': channel - }) + self._bus.post({"type": "join", "channel": channel}) continue if text.startswith("/announce "): - self._bus.post({ - 'type': 'announce', - 'text': text[10:] - }) + self._bus.post({"type": "announce", "text": text[10:]}) continue if self._channel is not None: - self._bus.post({ - 'channel': self._channel, - 'type': 'say', - 'text': text - }) + self._bus.post({"channel": self._channel, "type": "say", "text": text}) else: self._print("*** Need to /join a channel first") def _on_bus_message(self, message, data): - mtype = message['type'] - if mtype == 'welcome': - self._print("*** Welcome! Available channels:", repr(message['channels'])) - elif mtype == 'membership': - self._print("*** Joined", message['channel']) - self._print("- Members:\n\t" + "\n\t".join([f"{m['nick']} (connected from {m['address']})" for m in message['members']])) - for item in message['history']: + mtype = message["type"] + if mtype == "welcome": + self._print("*** Welcome! Available channels:", repr(message["channels"])) + elif mtype == "membership": + self._print("*** Joined", message["channel"]) + self._print( + "- Members:\n\t" + + "\n\t".join([f"{m['nick']} (connected from {m['address']})" for m in message["members"]]) + ) + for item in message["history"]: self._print(f"<{item['sender']}> {item['text']}") - elif mtype == 'join': - user = message['user'] + elif mtype == "join": + user = message["user"] self._print(f"👋 {user['nick']} ({user['address']}) joined {message['channel']}") - elif mtype == 'part': - user = message['user'] + elif mtype == "part": + user = message["user"] self._print(f"🚪 {user['nick']} ({user['address']}) left {message['channel']}") - elif mtype == 'chat': + elif mtype == "chat": self._print(f"<{message['sender']}> {message['text']}") - elif mtype == 'announce': + elif mtype == "announce": self._print(f"📣 <{message['sender']}> {message['text']}") else: self._print("Unhandled message:", message) @@ -102,7 +90,7 @@ class Application: sys.stdout.flush() -if __name__ == '__main__': +if __name__ == "__main__": nick = sys.argv[1] app = Application(nick) app.run() diff --git a/examples/portal_server.py b/examples/portal_server.py index 7dcc5dc..a99066b 100644 --- a/examples/portal_server.py +++ b/examples/portal_server.py @@ -1,10 +1,11 @@ -import frida -from frida_tools.application import Reactor import hashlib import hmac import json from pathlib import Path +from frida_tools.application import Reactor + +import frida ENABLE_CONTROL_INTERFACE = True @@ -13,16 +14,17 @@ class Application: def __init__(self): self._reactor = Reactor(run_until_return=self._process_input) - cluster_params = frida.EndpointParameters(address="unix:/Users/oleavr/src/cluster", - certificate="/Users/oleavr/src/identity2.pem", - authentication=('token', "wow-such-secret")) + cluster_params = frida.EndpointParameters( + address="unix:/Users/oleavr/src/cluster", + certificate="/Users/oleavr/src/identity2.pem", + authentication=("token", "wow-such-secret"), + ) if ENABLE_CONTROL_INTERFACE: www = Path(__file__).parent.resolve() / "web_client" / "dist" - control_params = frida.EndpointParameters(address="::1", - port=27042, - authentication=('callback', self._authenticate), - asset_root=www) + control_params = frida.EndpointParameters( + address="::1", port=27042, authentication=("callback", self._authenticate), asset_root=www + ) else: control_params = None @@ -33,15 +35,20 @@ class Application: self._nicks = set() self._channels = {} - service.on('node-connected', lambda *args: self._reactor.schedule(lambda: self._on_node_connected(*args))) - service.on('node-joined', lambda *args: self._reactor.schedule(lambda: self._on_node_joined(*args))) - service.on('node-left', lambda *args: self._reactor.schedule(lambda: self._on_node_left(*args))) - service.on('node-disconnected', lambda *args: self._reactor.schedule(lambda: self._on_node_disconnected(*args))) - service.on('controller-connected', lambda *args: self._reactor.schedule(lambda: self._on_controller_connected(*args))) - service.on('controller-disconnected', lambda *args: self._reactor.schedule(lambda: self._on_controller_disconnected(*args))) - service.on('authenticated', lambda *args: self._reactor.schedule(lambda: self._on_authenticated(*args))) - service.on('subscribe', lambda *args: self._reactor.schedule(lambda: self._on_subscribe(*args))) - service.on('message', lambda *args: self._reactor.schedule(lambda: self._on_message(*args))) + service.on("node-connected", lambda *args: self._reactor.schedule(lambda: self._on_node_connected(*args))) + service.on("node-joined", lambda *args: self._reactor.schedule(lambda: self._on_node_joined(*args))) + service.on("node-left", lambda *args: self._reactor.schedule(lambda: self._on_node_left(*args))) + service.on("node-disconnected", lambda *args: self._reactor.schedule(lambda: self._on_node_disconnected(*args))) + service.on( + "controller-connected", lambda *args: self._reactor.schedule(lambda: self._on_controller_connected(*args)) + ) + service.on( + "controller-disconnected", + lambda *args: self._reactor.schedule(lambda: self._on_controller_disconnected(*args)), + ) + service.on("authenticated", lambda *args: self._reactor.schedule(lambda: self._on_authenticated(*args))) + service.on("subscribe", lambda *args: self._reactor.schedule(lambda: self._on_subscribe(*args))) + service.on("message", lambda *args: self._reactor.schedule(lambda: self._on_message(*args))) def run(self): self._reactor.schedule(self._start) @@ -74,8 +81,8 @@ class Application: def _authenticate(self, raw_token): try: token = json.loads(raw_token) - nick = str(token['nick']) - secret = token['secret'].encode('utf-8') + nick = str(token["nick"]) + secret = token["secret"].encode("utf-8") except: raise ValueError("invalid request") @@ -85,7 +92,7 @@ class Application: raise ValueError("get outta here") return { - 'nick': nick, + "nick": nick, } def _on_node_connected(self, connection_id, remote_address): @@ -118,37 +125,30 @@ class Application: peer = self._peers.get(connection_id, None) if peer is None: return - peer.nick = self._acquire_nick(session_info['nick']) + peer.nick = self._acquire_nick(session_info["nick"]) def _on_subscribe(self, connection_id): print("on_subscribe()", connection_id) - self._service.post(connection_id, { - 'type': 'welcome', - 'channels': list(self._channels.keys()) - }) + self._service.post(connection_id, {"type": "welcome", "channels": list(self._channels.keys())}) def _on_message(self, connection_id, message, data): peer = self._peers[connection_id] - mtype = message['type'] - if mtype == 'join': - self._get_channel(message['channel']).add_member(peer) - elif mtype == 'part': - channel = self._channels.get(message['channel'], None) + mtype = message["type"] + if mtype == "join": + self._get_channel(message["channel"]).add_member(peer) + elif mtype == "part": + channel = self._channels.get(message["channel"], None) if channel is None: return channel.remove_member(peer) - elif mtype == 'say': - channel = self._channels.get(message['channel'], None) + elif mtype == "say": + channel = self._channels.get(message["channel"], None) if channel is None: return - channel.post(message['text'], peer) - elif mtype == 'announce': - self._service.broadcast({ - 'type': 'announce', - 'sender': peer.nick, - 'text': message['text'] - }) + channel.post(message["text"], peer) + elif mtype == "announce": + self._service.broadcast({"type": "announce", "sender": peer.nick, "text": message["text"]}) else: print("Unhandled message:", message) @@ -183,10 +183,7 @@ class Peer: self.memberships = set() def to_json(self): - return { - 'nick': self.nick, - 'address': self.remote_address[0] - } + return {"nick": self.nick, "address": self.remote_address[0]} class Channel: @@ -204,19 +201,18 @@ class Channel: peer.memberships.add(self) self.members.add(peer) - self._service.narrowcast(self.name, { - 'type': 'join', - 'channel': self.name, - 'user': peer.to_json() - }) + self._service.narrowcast(self.name, {"type": "join", "channel": self.name, "user": peer.to_json()}) self._service.tag(peer.connection_id, self.name) - self._service.post(peer.connection_id, { - 'type': 'membership', - 'channel': self.name, - 'members': [peer.to_json() for peer in self.members], - 'history': self.history - }) + self._service.post( + peer.connection_id, + { + "type": "membership", + "channel": self.name, + "members": [peer.to_json() for peer in self.members], + "history": self.history, + }, + ) def remove_member(self, peer): if self not in peer.memberships: @@ -226,21 +222,13 @@ class Channel: self.members.remove(peer) self._service.untag(peer.connection_id, self.name) - self._service.narrowcast(self.name, { - 'type': 'part', - 'channel': self.name, - 'user': peer.to_json() - }) + self._service.narrowcast(self.name, {"type": "part", "channel": self.name, "user": peer.to_json()}) def post(self, text, peer): if self not in peer.memberships: return - item = { - 'type': 'chat', - 'sender': peer.nick, - 'text': text - } + item = {"type": "chat", "sender": peer.nick, "text": text} self._service.narrowcast(self.name, item) @@ -250,6 +238,6 @@ class Channel: history.pop(0) -if __name__ == '__main__': +if __name__ == "__main__": app = Application() app.run() diff --git a/examples/query_system_parameters.py b/examples/query_system_parameters.py index 27d01e6..f5f38d9 100644 --- a/examples/query_system_parameters.py +++ b/examples/query_system_parameters.py @@ -1,9 +1,13 @@ -import frida from pprint import pformat + from pygments import highlight from pygments.formatters import Terminal256Formatter from pygments.lexers import PythonLexer +import frida print("Local parameters:", highlight(pformat(frida.query_system_parameters()), PythonLexer(), Terminal256Formatter())) -print("USB device parameters:", highlight(pformat(frida.get_usb_device().query_system_parameters()), PythonLexer(), Terminal256Formatter())) +print( + "USB device parameters:", + highlight(pformat(frida.get_usb_device().query_system_parameters()), PythonLexer(), Terminal256Formatter()), +) diff --git a/examples/rpc.py b/examples/rpc.py index 269b5be..5684a3b 100644 --- a/examples/rpc.py +++ b/examples/rpc.py @@ -1,8 +1,8 @@ import frida - session = frida.attach("Twitter") -script = session.create_script("""\ +script = session.create_script( + """\ rpc.exports = { hello: function () { return 'Hello'; @@ -11,7 +11,8 @@ rpc.exports = { oops; } }; -""") +""" +) script.load() api = script.exports print("api.hello() =>", api.hello()) diff --git a/examples/session_persist_timeout.py b/examples/session_persist_timeout.py index 02c614f..d749d11 100644 --- a/examples/session_persist_timeout.py +++ b/examples/session_persist_timeout.py @@ -1,6 +1,7 @@ -import frida from frida_tools.application import Reactor +import frida + class Application: def __init__(self): @@ -19,9 +20,10 @@ class Application: session = self._device.attach("hello2", persist_timeout=30) self._session = session - session.on('detached', lambda *args: self._reactor.schedule(lambda: self._on_detached(*args))) + session.on("detached", lambda *args: self._reactor.schedule(lambda: self._on_detached(*args))) - script = session.create_script(""" + script = session.create_script( + """ let _puts = null; Interceptor.attach(DebugSymbol.getFunctionByName('f'), { @@ -46,9 +48,10 @@ function puts(s) { } _puts(Memory.allocUtf8String(s)); } -""") +""" + ) self._script = script - script.on('message', lambda *args: self._reactor.schedule(lambda: self._on_message(*args))) + script.on("message", lambda *args: self._reactor.schedule(lambda: self._on_message(*args))) script.load() def _process_input(self, reactor): diff --git a/examples/snapshot.py b/examples/snapshot.py index f64d912..3abd3bb 100644 --- a/examples/snapshot.py +++ b/examples/snapshot.py @@ -1,6 +1,5 @@ import frida - embed_script = """ const button = { color: 'blue', @@ -28,9 +27,11 @@ session = frida.attach(0) snapshot = session.snapshot_script(embed_script, warmup_script=warmup_script, runtime=runtime) + def on_message(message, data): print("on_message:", message) + script = session.create_script(test_script, snapshot=snapshot, runtime=runtime) script.on("message", on_message) script.load() diff --git a/frida/__init__.py b/frida/__init__.py index dc86a1d..d1cc980 100644 --- a/frida/__init__.py +++ b/frida/__init__.py @@ -14,7 +14,6 @@ except Exception as ex: raise ex from . import core - __version__ = _frida.__version__ get_device_manager = core.get_device_manager @@ -70,15 +69,15 @@ def inject_library_blob(target, blob, entrypoint, data, **kwargs): def get_local_device(**kwargs): - return get_device_matching(lambda d: d.type == 'local', timeout=0, **kwargs) + return get_device_matching(lambda d: d.type == "local", timeout=0, **kwargs) def get_remote_device(**kwargs): - return get_device_matching(lambda d: d.type == 'remote', timeout=0, **kwargs) + return get_device_matching(lambda d: d.type == "remote", timeout=0, **kwargs) def get_usb_device(timeout=0, **kwargs): - return get_device_matching(lambda d: d.type == 'usb', timeout, **kwargs) + return get_device_matching(lambda d: d.type == "usb", timeout, **kwargs) def get_device(id, timeout=0, **kwargs): diff --git a/frida/core.py b/frida/core.py index 6754864..91f74b5 100644 --- a/frida/core.py +++ b/frida/core.py @@ -1,5 +1,5 @@ import fnmatch -from functools import wraps +import functools import json import numbers import sys @@ -8,8 +8,8 @@ import traceback import _frida - _device_manager = None + _Cancellable = _frida.Cancellable @@ -21,9 +21,9 @@ def get_device_manager(): def cancellable(f): - @wraps(f) + @functools.wraps(f) def wrapper(*args, **kwargs): - cancellable = kwargs.pop('cancellable', None) + cancellable = kwargs.pop("cancellable", None) if cancellable is not None: with cancellable: return f(*args, **kwargs) @@ -41,13 +41,13 @@ class DeviceManager: return repr(self._impl) def get_local_device(self, **kwargs): - return self.get_device_matching(lambda d: d.type == 'local', timeout=0, **kwargs) + return self.get_device_matching(lambda d: d.type == "local", timeout=0, **kwargs) def get_remote_device(self, **kwargs): - return self.get_device_matching(lambda d: d.type == 'remote', timeout=0, **kwargs) + return self.get_device_matching(lambda d: d.type == "remote", timeout=0, **kwargs) def get_usb_device(self, timeout=0, **kwargs): - return self.get_device_matching(lambda d: d.type == 'usb', timeout, **kwargs) + return self.get_device_matching(lambda d: d.type == "usb", timeout, **kwargs) def get_device(self, id, timeout=0, **kwargs): return self.get_device_matching(lambda d: d.id == id, timeout, **kwargs) @@ -117,7 +117,11 @@ class Device: @cancellable def get_process(self, process_name): process_name_lc = process_name.lower() - matching = [process for process in self._impl.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: @@ -204,7 +208,7 @@ class Bus: self._impl = impl self._on_message_callbacks = [] - impl.on('message', self._on_message) + impl.on("message", self._on_message) @cancellable def attach(self): @@ -215,13 +219,13 @@ class Bus: self._impl.post(raw_message, **kwargs) def on(self, signal, callback): - if signal == 'message': + if signal == "message": self._on_message_callbacks.append(callback) else: self._impl.on(signal, callback) def off(self, signal, callback): - if signal == 'message': + if signal == "message": self._on_message_callbacks.remove(callback) else: self._impl.off(signal, callback) @@ -307,8 +311,8 @@ class Script: self._next_request_id = 1 self._cond = threading.Condition() - impl.on('destroyed', self._on_destroyed) - impl.on('message', self._on_message) + impl.on("destroyed", self._on_destroyed) + impl.on("message", self._on_message) def __repr__(self): return repr(self._impl) @@ -342,13 +346,13 @@ class Script: self._impl.disable_debugger() def on(self, signal, callback): - if signal == 'message': + if signal == "message": self._on_message_callbacks.append(callback) else: self._impl.on(signal, callback) def off(self, signal, callback): - if signal == 'message': + if signal == "message": self._on_message_callbacks.remove(callback) else: self._impl.off(signal, callback) @@ -360,13 +364,13 @@ class Script: self._log_handler = handler def default_log_handler(self, level, text): - if level == 'info': + if level == "info": print(text, file=sys.stdout) else: print(text, file=sys.stderr) def list_exports(self): - return self._rpc_request('list') + return self._rpc_request("list") @cancellable def _rpc_request(self, *args): @@ -389,7 +393,7 @@ class Script: self._pending[request_id] = on_complete if not self.is_destroyed: - message = ['frida:rpc', request_id] + message = ["frida:rpc", request_id] message.extend(args) self.post(message) @@ -412,14 +416,14 @@ class Script: return result[1] def _on_rpc_message(self, request_id, operation, params, data): - if operation in ('ok', 'error'): + if operation in ("ok", "error"): callback = self._pending.pop(request_id, None) if callback is None: return value = None error = None - if operation == 'ok': + if operation == "ok": value = params[0] if data is None else data else: error = RPCException(*params[0:3]) @@ -438,18 +442,18 @@ class Script: if next_pending is None: break - next_pending(None, _frida.InvalidOperationError('script has been destroyed')) + next_pending(None, _frida.InvalidOperationError("script has been destroyed")) def _on_message(self, raw_message, data): message = json.loads(raw_message) - mtype = message['type'] - payload = message.get('payload', None) - if mtype == 'log': - level = message['level'] + mtype = message["type"] + payload = message.get("payload", None) + if mtype == "log": + level = message["level"] text = payload self._log_handler(level, text) - elif mtype == 'send' and isinstance(payload, list) and len(payload) > 0 and payload[0] == 'frida:rpc': + elif mtype == "send" and isinstance(payload, list) and len(payload) > 0 and payload[0] == "frida:rpc": request_id = payload[1] operation = payload[2] params = payload[3:] @@ -474,8 +478,10 @@ class ScriptExports: def __getattr__(self, name): script = self._script js_name = _to_camel_case(name) + def method(*args, **kwargs): - return script._rpc_request('call', js_name, args, **kwargs) + return script._rpc_request("call", js_name, args, **kwargs) + return method def __dir__(self): @@ -496,28 +502,28 @@ class EndpointParameters: kw = {} if address is not None: - kw['address'] = address + kw["address"] = address if port is not None: - kw['port'] = port + kw["port"] = port if certificate is not None: - kw['certificate'] = certificate + kw["certificate"] = certificate if origin is not None: - kw['origin'] = origin + kw["origin"] = origin if authentication is not None: (auth_scheme, auth_data) = authentication - if auth_scheme == 'token': - kw['auth_token'] = auth_data - elif auth_scheme == 'callback': - kw['auth_callback'] = make_auth_callback(auth_data) + if auth_scheme == "token": + kw["auth_token"] = auth_data + elif auth_scheme == "callback": + kw["auth_callback"] = make_auth_callback(auth_data) else: raise ValueError("invalid authentication scheme") if asset_root is not None: - kw['asset_root'] = str(asset_root) + kw["asset_root"] = str(asset_root) self._impl = _frida.EndpointParameters(**kw) @@ -534,8 +540,8 @@ class PortalService: self._on_authenticated_callbacks = [] self._on_message_callbacks = [] - impl.on('authenticated', self._on_authenticated) - impl.on('message', self._on_message) + impl.on("authenticated", self._on_authenticated) + impl.on("message", self._on_message) @cancellable def start(self): @@ -567,17 +573,17 @@ class PortalService: self._impl.untag(connection_id, tag) def on(self, signal, callback): - if signal == 'authenticated': + if signal == "authenticated": self._on_authenticated_callbacks.append(callback) - elif signal == 'message': + elif signal == "message": self._on_message_callbacks.append(callback) else: self._impl.on(signal, callback) def off(self, signal, callback): - if signal == 'authenticated': + if signal == "authenticated": self._on_authenticated_callbacks.remove(callback) - elif signal == 'message': + elif signal == "message": self._on_message_callbacks.remove(callback) else: self._impl.off(signal, callback) @@ -721,6 +727,7 @@ def make_auth_callback(callback): def authenticate(token): session_info = callback(token) return json.dumps(session_info) + return authenticate @@ -728,7 +735,7 @@ def _to_camel_case(name): result = "" uppercase_next = False for c in name: - if c == '_': + if c == "_": uppercase_next = True elif uppercase_next: result += c.upper() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..96fa357 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,6 @@ +[tool.black] +line-length = 120 + +[tool.isort] +profile = "black" +line_length = 120 diff --git a/setup.py b/setup.py index 0c3fcf3..e20129f 100755 --- a/setup.py +++ b/setup.py @@ -1,25 +1,24 @@ -import sys +import codecs +import hashlib import os import platform import re -import zipfile import shutil import struct -import codecs import subprocess -import hashlib +import sys +import zipfile from collections import namedtuple from functools import partial -from io import BytesIO -from urllib.request import urlopen -from urllib.parse import urljoin, urlparse, urlunparse from html.parser import HTMLParser +from io import BytesIO +from urllib.parse import urljoin, urlparse, urlunparse +from urllib.request import urlopen from setuptools import setup from setuptools.command.build_ext import build_ext from setuptools.extension import Extension - DEFAULT_INDEX_URL = "https://pypi.org/simple/" python_version = sys.version_info[0:2] @@ -28,20 +27,19 @@ package_dir = os.path.dirname(os.path.realpath(__file__)) pkg_info = os.path.join(package_dir, "PKG-INFO") in_source_package = os.path.isfile(pkg_info) if in_source_package: - with codecs.open(pkg_info, "r", 'utf-8') as f: + with codecs.open(pkg_info, "r", "utf-8") as f: version_line = [line.rstrip("\r") for line in f.read().split("\n") if line.startswith("Version: ")][0] frida_version = version_line[9:] long_description = None else: - frida_version = os.environ.get('FRIDA_VERSION', None) - long_description = codecs.open(os.path.join(package_dir, "README.md"), "r", 'utf-8').read() - frida_extension = os.environ.get('FRIDA_EXTENSION', None) + frida_version = os.environ.get("FRIDA_VERSION", None) + long_description = codecs.open(os.path.join(package_dir, "README.md"), "r", "utf-8").read() + frida_extension = os.environ.get("FRIDA_EXTENSION", None) index_url_pip_configs = ("global.index-url", "global.extra-index-url") Tag = namedtuple("Tag", ["tagname", "attrs"]) -ParsedUrlInfo = namedtuple("ParsedUrlInfo", - ["url", "filename", "major", "minor", "micro"]) +ParsedUrlInfo = namedtuple("ParsedUrlInfo", ["url", "filename", "major", "minor", "micro"]) class FridaPrebuiltExt(build_ext): @@ -56,31 +54,36 @@ class FridaPrebuiltExt(build_ext): if in_source_package: system = platform.system() - arch = struct.calcsize('P') * 8 - if system == 'Windows': + arch = struct.calcsize("P") * 8 + if system == "Windows": os_version = "win-amd64" if arch == 64 else "win32" - elif system == 'Darwin': - if platform.machine() == 'x86_64': + elif system == "Darwin": + if platform.machine() == "x86_64": os_version = "macosx-10.9-x86_64" else: os_version = "macosx-11.0-arm64" - elif system == 'Linux': - os_name = 'android' if subprocess.check_output(["uname", "-o"]).decode('utf-8').rstrip() == 'Android' else 'linux' + elif system == "Linux": + os_name = ( + "android" + if subprocess.check_output(["uname", "-o"]).decode("utf-8").rstrip() == "Android" + else "linux" + ) machine = platform.machine() if machine == "" or "86" in machine: arch_name = "x86_64" if arch == 64 else "i686" - elif os_name == 'android' and machine.startswith("armv"): - arch_name = 'armv7l' + elif os_name == "android" and machine.startswith("armv"): + arch_name = "armv7l" else: arch_name = machine os_version = f"{os_name}-{arch_name}" - elif system == 'FreeBSD': + elif system == "FreeBSD": os_version = "freebsd-" + platform.machine() else: raise NotImplementedError("unsupported OS") egg_path = os.path.expanduser( - f"~{os.sep}frida-{frida_version}-py{python_version[0]}.{python_version[1]}-{os_version}.egg") + f"~{os.sep}frida-{frida_version}-py{python_version[0]}.{python_version[1]}-{os_version}.egg" + ) print("looking for prebuilt extension in home directory, i.e.", egg_path) try: @@ -111,8 +114,9 @@ class FridaPrebuiltExt(build_ext): parser.feed(links_html) if len(parser.urls) == 0: - raise NotImplementedError("could not find prebuilt Frida extension; " - "prebuilds only provided for Python 3.4+") + raise NotImplementedError( + "could not find prebuilt Frida extension; " "prebuilds only provided for Python 3.4+" + ) url = parser.urls[0] egg_url = urljoin(frida_url, url.url) @@ -137,12 +141,14 @@ class FridaPrebuiltExt(build_ext): egg_zip = zipfile.ZipFile(egg_file) extension_member = [info for info in egg_zip.infolist() if info.filename.endswith(target_extension)][0] extension_data = egg_zip.read(extension_member) - if system == 'Windows': + if system == "Windows": trailer = b"\x00" if python_version[1] >= 10 else b"\x00\x00" - extension_data = re.sub(b"python[3-9][0-9][0-9]\\.dll\x00", - "python{}{}.dll".format(*python_version).encode('utf-8') + trailer, - extension_data) - with open(target, 'wb') as f: + extension_data = re.sub( + b"python[3-9][0-9][0-9]\\.dll\x00", + "python{}{}.dll".format(*python_version).encode("utf-8") + trailer, + extension_data, + ) + with open(target, "wb") as f: f.write(extension_data) else: shutil.copyfile(frida_extension, target) @@ -172,8 +178,9 @@ def get_index_url(): def get_index_url_from_pip(config_name): assert config_name in index_url_pip_configs - return subprocess.check_output([sys.executable, "-m", "pip", "config", "get", config_name], - stderr=subprocess.PIPE).decode("utf-8") + return subprocess.check_output( + [sys.executable, "-m", "pip", "config", "get", config_name], stderr=subprocess.PIPE + ).decode("utf-8") def normalize_url(url): @@ -181,18 +188,24 @@ def normalize_url(url): path = parse_result.path if not path.endswith("/"): path += "/" - return urlunparse(( - parse_result.scheme, parse_result.netloc, path, - parse_result.params, parse_result.query, parse_result.fragment, - )) + return urlunparse( + ( + parse_result.scheme, + parse_result.netloc, + path, + parse_result.params, + parse_result.query, + parse_result.fragment, + ) + ) class PEP503PageParser(HTMLParser): def __init__(self, name, version, os_version): HTMLParser.__init__(self) - filename_pattern = ( - r"^{}\-{}\-py(?P\d+)\.(?P\d+)(\.(?P\d+))?-{}.egg$" - ).format(*map(re.escape, [name, version, os_version])) + filename_pattern = (r"^{}\-{}\-py(?P\d+)\.(?P\d+)(\.(?P\d+))?-{}.egg$").format( + *map(re.escape, [name, version, os_version]) + ) self._filename_pattern = re.compile(filename_pattern) def reset(self): @@ -213,21 +226,18 @@ class PEP503PageParser(HTMLParser): self._path.pop() def handle_data(self, data): - if not (len(self._path) > 0 - and self._path[-1].tagname == "a" - and self._path[-1].attrs.get("href")): + if not (len(self._path) > 0 and self._path[-1].tagname == "a" and self._path[-1].attrs.get("href")): return match = self._filename_pattern.match(data) if match is not None: - self.urls.append(ParsedUrlInfo( - self._path[-1].attrs["href"], - data, - *map( - lambda g: int(g) if g else None, - map(match.group, ["major", "minor", "micro"]) + self.urls.append( + ParsedUrlInfo( + self._path[-1].attrs["href"], + data, + *map(lambda g: int(g) if g else None, map(match.group, ["major", "minor", "micro"])), ) - )) + ) def check_pep503_hash(bytes_io, url): @@ -286,12 +296,10 @@ if __name__ == "__main__": "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: JavaScript", "Topic :: Software Development :: Debuggers", - "Topic :: Software Development :: Libraries :: Python Modules" + "Topic :: Software Development :: Libraries :: Python Modules", ], - packages=['frida'], - ext_modules=[Extension('_frida', [])], - cmdclass={ - 'build_ext': FridaPrebuiltExt - }, - zip_safe=False + packages=["frida"], + ext_modules=[Extension("_frida", [])], + cmdclass={"build_ext": FridaPrebuiltExt}, + zip_safe=False, ) diff --git a/tests/data/__init__.py b/tests/data/__init__.py index 45c35e1..894fa90 100644 --- a/tests/data/__init__.py +++ b/tests/data/__init__.py @@ -2,17 +2,16 @@ import os import platform import sys - system = platform.system() -if system == 'Windows': +if system == "Windows": target_program = r"C:\Windows\notepad.exe" -elif system == 'Darwin': +elif system == "Darwin": target_program = os.path.join(os.path.dirname(__file__), "unixvictim-macos") -elif system == 'Linux' and platform.machine() == 'x86_64': - arch = 'x86_64' if sys.maxsize > 2**32 else 'x86' +elif system == "Linux" and platform.machine() == "x86_64": + arch = "x86_64" if sys.maxsize > 2**32 else "x86" target_program = os.path.join(os.path.dirname(__file__), "unixvictim-" + system.lower() + "-" + arch) else: target_program = "/bin/cat" -__all__ = ['target_program'] +__all__ = ["target_program"] diff --git a/tests/test_core.py b/tests/test_core.py index 954c628..720fbaa 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -12,27 +12,31 @@ class TestCore(unittest.TestCase): self.assertTrue(len(devices) > 0) def test_get_existing_device(self): - device = frida.get_device_matching(lambda d: d.id == 'local') + device = frida.get_device_matching(lambda d: d.id == "local") self.assertEqual(device.name, "Local System") - device = frida.get_device_manager().get_device_matching(lambda d: d.id == 'local') + device = frida.get_device_manager().get_device_matching(lambda d: d.id == "local") self.assertEqual(device.name, "Local System") def test_get_nonexistent_device(self): def get_nonexistent(): - frida.get_device_manager().get_device_matching(lambda device: device.type == 'lol') + frida.get_device_manager().get_device_matching(lambda device: device.type == "lol") + self.assertRaisesMatching(frida.InvalidArgumentError, "device not found", get_nonexistent) def test_wait_for_nonexistent_device(self): def wait_for_nonexistent(): - frida.get_device_manager().get_device_matching(lambda device: device.type == 'lol', timeout=0.1) + frida.get_device_manager().get_device_matching(lambda device: device.type == "lol", timeout=0.1) + self.assertRaisesMatching(frida.InvalidArgumentError, "device not found", wait_for_nonexistent) def test_cancel_wait_for_nonexistent_device(self): cancellable = frida.Cancellable() def wait_for_nonexistent(): - frida.get_device_manager().get_device_matching(lambda device: device.type == 'lol', timeout=-1, cancellable=cancellable) + frida.get_device_manager().get_device_matching( + lambda device: device.type == "lol", timeout=-1, cancellable=cancellable + ) def cancel_after_100ms(): time.sleep(0.1) @@ -46,5 +50,5 @@ class TestCore(unittest.TestCase): m(exception, regex, operation) -if __name__ == '__main__': +if __name__ == "__main__": unittest.main() diff --git a/tests/test_pep503_page_parser.py b/tests/test_pep503_page_parser.py index f50b578..c85d692 100644 --- a/tests/test_pep503_page_parser.py +++ b/tests/test_pep503_page_parser.py @@ -3,47 +3,42 @@ import unittest import setup - py_major_version = sys.version_info[0] htmls = [] cases = [] html = ( - '\n\n \n Links for frida\n ' + "\n\n \n Links for frida\n " ' \n \n

Links for frida

\n frida-1.4.1-py2.6-macosx-10.9-' 'intel.egg
\nfrida-9.1.9.t' - 'ar.gz
\n \n\n' + "ar.gz
\n \n\n" ) htmls.append(html) -cases.extend([ - ( - setup.PEP503PageParser("frida", "15.1.1", "win-amd64"), - html, - [] - ), - ( - setup.PEP503PageParser("frida", "1.4.1", "macosx-10.9-intel"), - html, - [ - setup.ParsedUrlInfo( - url='../../packages/5d/80/3b140c5998df9d81e40169f188a2347b6c705156a2b556ff308e2f8b7e0a/frida-1.4.1-py2.6-macosx-10.9-intel.egg#sha256=eef92210084ef083b34f8972078550c6ef45255e444905f95495792c7f709546', - filename='frida-1.4.1-py2.6-macosx-10.9-intel.egg', - major=2, minor=6, micro=None - ) - ] - ), - ( - setup.PEP503PageParser("frida", "1.4.1", "macosx-11.0-arm64"), - html, - [] - ), -]) +cases.extend( + [ + (setup.PEP503PageParser("frida", "15.1.1", "win-amd64"), html, []), + ( + setup.PEP503PageParser("frida", "1.4.1", "macosx-10.9-intel"), + html, + [ + setup.ParsedUrlInfo( + url="../../packages/5d/80/3b140c5998df9d81e40169f188a2347b6c705156a2b556ff308e2f8b7e0a/frida-1.4.1-py2.6-macosx-10.9-intel.egg#sha256=eef92210084ef083b34f8972078550c6ef45255e444905f95495792c7f709546", + filename="frida-1.4.1-py2.6-macosx-10.9-intel.egg", + major=2, + minor=6, + micro=None, + ) + ], + ), + (setup.PEP503PageParser("frida", "1.4.1", "macosx-11.0-arm64"), html, []), + ] +) html = ( 'frida-15.0.7-py2.7-linux-i686.egg
' @@ -80,121 +75,137 @@ html = ( 'frida-15.0.8.tar.gz
' ) htmls.append(html) -cases.extend([ - ( - setup.PEP503PageParser("frida", "15.1.0", "win-amd64"), - html, - [] - ), - ( - setup.PEP503PageParser("frida", "15.0.7", "win-amd64"), - html, - [ - setup.ParsedUrlInfo( - url='../../packages/e0/5c/b45c8f27482d81179eb640726b703f95c624cc4f32ae3ed3f8bc858ae5d9/frida-15.0.7-py2.7-win-amd64.egg#sha256=eb696528b9c19f1895123e731b094363a87a4412d7ea4fcb54ef71841f7b3c1e', - filename='frida-15.0.7-py2.7-win-amd64.egg', - major=2, minor=7, micro=None - ), - setup.ParsedUrlInfo( - url='../../packages/77/34/6ebaea697f3df72818e60c6494a716c51f7f13b3da323598c1711d21779c/frida-15.0.7-py3.8-win-amd64.egg#sha256=a9964cc6dd4e3ea71c42b1800c79571c670905dc82cd769302b066499fff7bf4', - filename='frida-15.0.7-py3.8-win-amd64.egg', - major=3, minor=8, micro=None - ), - ] - ), - ( - setup.PEP503PageParser("frida", "15.0.7", "linux-i686"), - html, - [ - setup.ParsedUrlInfo( - url='../../packages/e3/21/da75f6207f76750799d68938707a74d46512e666293eb550247bf5314613/frida-15.0.7-py2.7-linux-i686.egg#sha256=444246bad3b2222efec301e96c2d6ac5da039d41acd655f6d5b6e548637cae09', - filename='frida-15.0.7-py2.7-linux-i686.egg', - major=2, minor=7, micro=None - ), - setup.ParsedUrlInfo( - url='../../packages/19/d3/a4a1980005e232399575aeb2ae973d2087a94ec7dbaf6d7a481612979fc7/frida-15.0.7-py3.8-linux-i686.egg#sha256=ed922ec0258e95f39b4004066b72fb48546041d28602e55d44ca12effa80e8bf', - filename='frida-15.0.7-py3.8-linux-i686.egg', - major=3, minor=8, micro=None - ), - ] - ), - ( - setup.PEP503PageParser("frida", "15.0.8", "linux-x86_64"), - html, - [ - setup.ParsedUrlInfo( - url='../../packages/64/4a/1e1735a8c2f606c953cccfb9d7086c15d19b5151ebd6e0cbcab2e817d6e2/frida-15.0.8-py2.7-linux-x86_64.egg#sha256=e5b29da8394ef5643fc42877856859d544cd2aba0a874a4a700f2ce4521d9780', - filename='frida-15.0.8-py2.7-linux-x86_64.egg', - major=2, minor=7, micro=None - ), - setup.ParsedUrlInfo( - url='../../packages/0b/20/11101c2cc053bbe3695c8778ffb239e49c0bc24066257bc3246ef67770d9/frida-15.0.8-py3.8-linux-x86_64.egg#sha256=6b3f42225c22a1f149107f963abe9f7b5f32eb4915fe8fa8286e5657a7b6c789', - filename='frida-15.0.8-py3.8-linux-x86_64.egg', - major=3, minor=8, micro=None - ), - ] - ), - ( - setup.PEP503PageParser("frida", "15.0.7", "linux-amd64"), - html, - [] - ), - ( - setup.PEP503PageParser("frida", "15.0.8", "macosx-11.0-fat64"), - html, - [ - setup.ParsedUrlInfo( - url='../../packages/d1/20/a65170d6a898541839acb03a16d1dd26499928c937350078765fe1e4beb3/frida-15.0.8-py2.7-macosx-11.0-fat64.egg#sha256=f9e58ff7f6d53640a991d3e77711b0095927103d7bdfef55268b58091938f72e', - filename='frida-15.0.8-py2.7-macosx-11.0-fat64.egg', - major=2, minor=7, micro=None - ), - ] - ), -]) +cases.extend( + [ + (setup.PEP503PageParser("frida", "15.1.0", "win-amd64"), html, []), + ( + setup.PEP503PageParser("frida", "15.0.7", "win-amd64"), + html, + [ + setup.ParsedUrlInfo( + url="../../packages/e0/5c/b45c8f27482d81179eb640726b703f95c624cc4f32ae3ed3f8bc858ae5d9/frida-15.0.7-py2.7-win-amd64.egg#sha256=eb696528b9c19f1895123e731b094363a87a4412d7ea4fcb54ef71841f7b3c1e", + filename="frida-15.0.7-py2.7-win-amd64.egg", + major=2, + minor=7, + micro=None, + ), + setup.ParsedUrlInfo( + url="../../packages/77/34/6ebaea697f3df72818e60c6494a716c51f7f13b3da323598c1711d21779c/frida-15.0.7-py3.8-win-amd64.egg#sha256=a9964cc6dd4e3ea71c42b1800c79571c670905dc82cd769302b066499fff7bf4", + filename="frida-15.0.7-py3.8-win-amd64.egg", + major=3, + minor=8, + micro=None, + ), + ], + ), + ( + setup.PEP503PageParser("frida", "15.0.7", "linux-i686"), + html, + [ + setup.ParsedUrlInfo( + url="../../packages/e3/21/da75f6207f76750799d68938707a74d46512e666293eb550247bf5314613/frida-15.0.7-py2.7-linux-i686.egg#sha256=444246bad3b2222efec301e96c2d6ac5da039d41acd655f6d5b6e548637cae09", + filename="frida-15.0.7-py2.7-linux-i686.egg", + major=2, + minor=7, + micro=None, + ), + setup.ParsedUrlInfo( + url="../../packages/19/d3/a4a1980005e232399575aeb2ae973d2087a94ec7dbaf6d7a481612979fc7/frida-15.0.7-py3.8-linux-i686.egg#sha256=ed922ec0258e95f39b4004066b72fb48546041d28602e55d44ca12effa80e8bf", + filename="frida-15.0.7-py3.8-linux-i686.egg", + major=3, + minor=8, + micro=None, + ), + ], + ), + ( + setup.PEP503PageParser("frida", "15.0.8", "linux-x86_64"), + html, + [ + setup.ParsedUrlInfo( + url="../../packages/64/4a/1e1735a8c2f606c953cccfb9d7086c15d19b5151ebd6e0cbcab2e817d6e2/frida-15.0.8-py2.7-linux-x86_64.egg#sha256=e5b29da8394ef5643fc42877856859d544cd2aba0a874a4a700f2ce4521d9780", + filename="frida-15.0.8-py2.7-linux-x86_64.egg", + major=2, + minor=7, + micro=None, + ), + setup.ParsedUrlInfo( + url="../../packages/0b/20/11101c2cc053bbe3695c8778ffb239e49c0bc24066257bc3246ef67770d9/frida-15.0.8-py3.8-linux-x86_64.egg#sha256=6b3f42225c22a1f149107f963abe9f7b5f32eb4915fe8fa8286e5657a7b6c789", + filename="frida-15.0.8-py3.8-linux-x86_64.egg", + major=3, + minor=8, + micro=None, + ), + ], + ), + (setup.PEP503PageParser("frida", "15.0.7", "linux-amd64"), html, []), + ( + setup.PEP503PageParser("frida", "15.0.8", "macosx-11.0-fat64"), + html, + [ + setup.ParsedUrlInfo( + url="../../packages/d1/20/a65170d6a898541839acb03a16d1dd26499928c937350078765fe1e4beb3/frida-15.0.8-py2.7-macosx-11.0-fat64.egg#sha256=f9e58ff7f6d53640a991d3e77711b0095927103d7bdfef55268b58091938f72e", + filename="frida-15.0.8-py2.7-macosx-11.0-fat64.egg", + major=2, + minor=7, + micro=None, + ), + ], + ), + ] +) html = ( - '

frida-15.1.1-py3.8-linux-x86_64.egg

' + "

frida-15.1.1-py3.8-linux-x86_64.egg

" 'frida-15.1.1-py3.8-' - 'linux-x86_64.egg
' + "linux-x86_64.egg
" ) htmls.append(html) -cases.extend([ - ( - setup.PEP503PageParser("frida", "15.1.1", "linux-x86_64"), - html, - [ - setup.ParsedUrlInfo( - url='../../packages/e4/c1/82e361bbaa535b334f5b1b432b4573a7871fa973edeb3aab9dbb6b3b4cdc/frida-15.1.1-py3.8-linux-x86_64.egg#sha256=505f4ffa34cc7d68664fcd00d469f5d832e6778800d112aadb8a13692f984b40', - filename='frida-15.1.1-py3.8-linux-x86_64.egg', - major=3, minor=8, micro=None - ), - ] - ), -]) +cases.extend( + [ + ( + setup.PEP503PageParser("frida", "15.1.1", "linux-x86_64"), + html, + [ + setup.ParsedUrlInfo( + url="../../packages/e4/c1/82e361bbaa535b334f5b1b432b4573a7871fa973edeb3aab9dbb6b3b4cdc/frida-15.1.1-py3.8-linux-x86_64.egg#sha256=505f4ffa34cc7d68664fcd00d469f5d832e6778800d112aadb8a13692f984b40", + filename="frida-15.1.1-py3.8-linux-x86_64.egg", + major=3, + minor=8, + micro=None, + ), + ], + ), + ] +) html = ( 'frida-15.0.1-py3.8-android-aarch64.egg' 'frida-15.0.1-py3' - '.8-android-aarch64.egg
' + ".8-android-aarch64.egg
" ) htmls.append(html) -cases.extend([ - ( - setup.PEP503PageParser("frida", "15.0.1", "android-aarch64"), - html, - [ - setup.ParsedUrlInfo( - url='../../packages/3e/80/78fa3ed5fd636b606dc06157069b37eb677652cd985739cde35a86d7a362/frida-15.0.1-py3.8-android-aarch64.egg#sha256=d44bc341590dd8cf2623089b54aa16d697536fb016fde0ecd6df5262723c652b', - filename='frida-15.0.1-py3.8-android-aarch64.egg', - major=3, minor=8, micro=None - ), - ] - ), -]) +cases.extend( + [ + ( + setup.PEP503PageParser("frida", "15.0.1", "android-aarch64"), + html, + [ + setup.ParsedUrlInfo( + url="../../packages/3e/80/78fa3ed5fd636b606dc06157069b37eb677652cd985739cde35a86d7a362/frida-15.0.1-py3.8-android-aarch64.egg#sha256=d44bc341590dd8cf2623089b54aa16d697536fb016fde0ecd6df5262723c652b", + filename="frida-15.0.1-py3.8-android-aarch64.egg", + major=3, + minor=8, + micro=None, + ), + ], + ), + ] +) class TestPEP503PageParser(unittest.TestCase): diff --git a/tests/test_rpc.py b/tests/test_rpc.py index f8a8dfa..99f0812 100644 --- a/tests/test_rpc.py +++ b/tests/test_rpc.py @@ -27,7 +27,9 @@ class TestRpc(unittest.TestCase): cls.target.wait() def test_basics(self): - script = self.session.create_script(name="test-rpc", source="""\ + script = self.session.create_script( + name="test-rpc", + source="""\ rpc.exports = { add: function (a, b) { var result = a + b; @@ -43,21 +45,24 @@ rpc.exports = { return Memory.readByteArray(buf, 2); } }; -""") +""", + ) script.load() self.assertEqual(script.exports.add(2, 3), 5) self.assertEqual(script.exports.sub(5, 3), 2) self.assertRaises(Exception, lambda: script.exports.add(1, -2)) - self.assertListEqual([x for x in iterbytes(script.exports.speak())], - [0x59, 0x6f]) + self.assertListEqual([x for x in iterbytes(script.exports.speak())], [0x59, 0x6F]) def test_post_failure(self): - script = self.session.create_script(name="test-rpc", source="""\ + script = self.session.create_script( + name="test-rpc", + source="""\ rpc.exports = { init: function () { }, }; -""") +""", + ) script.load() agent = script.exports @@ -66,13 +71,16 @@ rpc.exports = { self.assertEqual(script._pending, {}) def test_unload_mid_request(self): - script = self.session.create_script(name="test-rpc", source="""\ + script = self.session.create_script( + name="test-rpc", + source="""\ rpc.exports = { waitForever: function () { return new Promise(function () {}); }, }; -""") +""", + ) script.load() agent = script.exports @@ -85,13 +93,16 @@ rpc.exports = { self.assertEqual(script._pending, {}) def test_detach_mid_request(self): - script = self.session.create_script(name="test-rpc", source="""\ + script = self.session.create_script( + name="test-rpc", + source="""\ rpc.exports = { waitForever: function () { return new Promise(function () {}); }, }; -""") +""", + ) script.load() agent = script.exports @@ -104,13 +115,16 @@ rpc.exports = { self.assertEqual(script._pending, {}) def test_cancellation_mid_request(self): - script = self.session.create_script(name="test-rpc", source="""\ + script = self.session.create_script( + name="test-rpc", + source="""\ rpc.exports = { waitForever: function () { return new Promise(function () {}); }, }; -""") +""", + ) script.load() agent = script.exports @@ -126,6 +140,7 @@ rpc.exports = { def call_wait_forever_with_cancellable(): with cancellable: agent.wait_forever() + cancellable = frida.Cancellable() threading.Thread(target=cancel_after_100ms).start() self.assertRaisesOperationCancelled(call_wait_forever_with_cancellable) @@ -145,9 +160,10 @@ rpc.exports = { 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__': +if __name__ == "__main__": unittest.main()