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

Automatically format code with Black and Isort

Browse Source
This commit is contained in:
Yotam Nachum authored and Orip committed 2022-09-12 20:59:52 +03:00
1 parent 32df9f05b2
commit fc00ac2b75
30 files changed
+522 -465
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

examples/bytecode.py
+5 -3
View File
@@ -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)
examples/channels.py
-1
View File
@@ -1,6 +1,5 @@
import frida
device = frida.get_usb_device()
channel = device.open_channel("tcp:21")
examples/child_gating.py
+7 -4
View File
@@ -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()
examples/cpushark/AppDelegate.py
+1
View File
@@ -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()
examples/cpushark/Capture.py
+41 -45
View File
@@ -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);
examples/cpushark/CpuShark.py
+2
View File
@@ -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()
examples/cpushark/MainWindowController.py
+4 -3
View File
@@ -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)
examples/cpushark/ProcessList.py
+2 -2
View File
@@ -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)
examples/cpushark/setup.py
+8 -8
View File
@@ -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"],
)
examples/crash_reporting.py
+4 -2
View File
@@ -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()
examples/detached.py
+6 -3
View File
@@ -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()
examples/enumerate_applications.py
+9 -6
View File
@@ -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})')
examples/enumerate_processes.py
+11 -6
View File
@@ -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()})'
)
examples/get_frontmost_application.py
+8 -5
View File
@@ -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")
examples/inject_library/inject_blob.py
+1
View File
@@ -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()
examples/inject_library/inject_file.py
+1
View File
@@ -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()
examples/portal_client.py
+28 -40
View File
@@ -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()
examples/portal_server.py
+54 -66
View File
@@ -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()
examples/query_system_parameters.py
+6 -2
View File
@@ -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()),
)
examples/rpc.py
+4 -3
View File
@@ -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())
examples/session_persist_timeout.py
+8 -5
View File
@@ -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):
examples/snapshot.py
+2 -1
View File
@@ -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()
frida/__init__.py
+3 -4
View File
@@ -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):
frida/core.py
+50 -43
View File
@@ -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()
pyproject.toml
+6
View File
@@ -0,0 +1,6 @@
[tool.black]
line-length = 120
[tool.isort]
profile = "black"
line_length = 120
setup.py
+65 -57
View File
@@ -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<major>\d+)\.(?P<minor>\d+)(\.(?P<micro>\d+))?-{}.egg$"
).format(*map(re.escape, [name, version, os_version]))
filename_pattern = (r"^{}\-{}\-py(?P<major>\d+)\.(?P<minor>\d+)(\.(?P<micro>\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,
)
tests/data/__init__.py
+5 -6
View File
@@ -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"]
tests/test_core.py
+10 -6
View File
@@ -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()
tests/test_pep503_page_parser.py
+142 -131
View File
@@ -3,47 +3,42 @@ import unittest
import setup
py_major_version = sys.version_info[0]
htmls = []
cases = []
html = (
'<!DOCTYPE html>\n<html>\n <head>\n <title>Links for frida</title>\n '
"<!DOCTYPE html>\n<html>\n <head>\n <title>Links for frida</title>\n "
' </head>\n <body>\n <h1>Links for frida</h1>\n <a href="../../pac'
'kages/5d/80/3b140c5998df9d81e40169f188a2347b6c705156a2b556ff308e2f8b7e0a'
'/frida-1.4.1-py2.6-macosx-10.9-intel.egg#sha256=eef92210084ef083b34f8972'
"kages/5d/80/3b140c5998df9d81e40169f188a2347b6c705156a2b556ff308e2f8b7e0a"
"/frida-1.4.1-py2.6-macosx-10.9-intel.egg#sha256=eef92210084ef083b34f8972"
'078550c6ef45255e444905f95495792c7f709546">frida-1.4.1-py2.6-macosx-10.9-'
'intel.egg</a><br/>\n<a href="../../packages/4e/ca/ee40ef1d5013300a77152f'
'f0687caedc3b5ea1f786bf3e0b778de5fc0b8a/frida-9.1.9.tar.gz#sha256=d215884'
"f0687caedc3b5ea1f786bf3e0b778de5fc0b8a/frida-9.1.9.tar.gz#sha256=d215884"
'4cc20cd3e2f8d2cd95f90449f0c27d051d9868706c1eda5d357eb86d7">frida-9.1.9.t'
'ar.gz</a><br/>\n </body>\n</html>\n<!--SERIAL 11362971-->'
"ar.gz</a><br/>\n </body>\n</html>\n<!--SERIAL 11362971-->"
)
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 = (
'<a href="../../packages/e3/21/da75f6207f76750799d68938707a74d46512e666293eb550247bf5314613/frida-15.0.7-py2.7-linux-i686.egg#sha256=444246bad3b2222efec301e96c2d6ac5da039d41acd655f6d5b6e548637cae09">frida-15.0.7-py2.7-linux-i686.egg</a><br/>'
@@ -80,121 +75,137 @@ html = (
'<a href="../../packages/e4/0f/9954d94b174ba703b7018ae01c5e37189715a7b1616a2341794ccbefe834/frida-15.0.8.tar.gz#sha256=de2df2924770601ce39cdc992fa3690b4a0891d614a515cad03bc1b94e762ff1">frida-15.0.8.tar.gz</a><br/>'
)
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 = (
'<h3>frida-15.1.1-py3.8-linux-x86_64.egg</h3>'
"<h3>frida-15.1.1-py3.8-linux-x86_64.egg</h3>"
'<a href="../../packages/e4/c1/82e361bbaa535b334f5b1b432b4573a7871fa973ede'
'b3aab9dbb6b3b4cdc/frida-15.1.1-py3.8-linux-x86_64.egg#sha256=505f4ffa34cc'
"b3aab9dbb6b3b4cdc/frida-15.1.1-py3.8-linux-x86_64.egg#sha256=505f4ffa34cc"
'7d68664fcd00d469f5d832e6778800d112aadb8a13692f984b40">frida-15.1.1-py3.8-'
'linux-x86_64.egg</a><br/>'
"linux-x86_64.egg</a><br/>"
)
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 = (
'<a role="button" tabindex="0">frida-15.0.1-py3.8-android-aarch64.egg</a>'
'<a href="../../packages/3e/80/78fa3ed5fd636b606dc06157069b37eb677652cd985'
'739cde35a86d7a362/frida-15.0.1-py3.8-android-aarch64.egg#sha256=d44bc3415'
"739cde35a86d7a362/frida-15.0.1-py3.8-android-aarch64.egg#sha256=d44bc3415"
'90dd8cf2623089b54aa16d697536fb016fde0ecd6df5262723c652b">frida-15.0.1-py3'
'.8-android-aarch64.egg</a><br/>'
".8-android-aarch64.egg</a><br/>"
)
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):
tests/test_rpc.py
+29 -13
View File
@@ -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()