diff --git a/.github/workflows/code-style.yml b/.github/workflows/code-style.yml new file mode 100644 index 0000000..3598378 --- /dev/null +++ b/.github/workflows/code-style.yml @@ -0,0 +1,16 @@ +name: code-style +on: [push, pull_request] +jobs: + black: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v3 + - uses: psf/black@stable + isort: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v3 + - uses: actions/setup-python@v4 + with: + python-version: 3.8 + - uses: jamescurtin/isort-action@master diff --git a/_frida/__init__.pyi b/_frida/__init__.pyi new file mode 100644 index 0000000..eb256be --- /dev/null +++ b/_frida/__init__.pyi @@ -0,0 +1,658 @@ +from typing import Any, Callable, ClassVar, Dict, List, Optional, Sequence, Tuple, Union + +# Exceptions +class AddressInUseError(Exception): ... +class ExecutableNotFoundError(Exception): ... +class ExecutableNotSupportedError(Exception): ... +class ServerNotRunningError(Exception): ... +class TimedOutError(Exception): ... +class TransportError(Exception): ... +class ProcessNotFoundError(Exception): ... +class ProcessNotRespondingError(Exception): ... +class ProtocolError(Exception): ... +class InvalidArgumentError(Exception): ... +class InvalidOperationError(Exception): ... +class NotSupportedError(Exception): ... +class OperationCancelledError(Exception): ... +class PermissionDeniedError(Exception): ... + +class Object: + def __init__(self, *args: Any, **kwargs: Any) -> None: ... + def on(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Add a signal handler. + """ + ... + def off(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Remove a signal handler. + """ + ... + +class Application(Object): + @property + def identifier(self) -> str: + """ + Application identifier. + """ + ... + @property + def name(self) -> str: + """ + Human-readable application name. + """ + ... + @property + def parameters(self) -> Dict[str, Any]: + """ + Parameters. + """ + ... + @property + def pid(self) -> int: + """ + Process ID, or 0 if not running. + """ + ... + +class Bus(Object): + def attach(self) -> None: + """ + Attach to the bus. + """ + ... + def post(self, message: str, data: Optional[Union[bytes, str]]) -> None: + """ + Post a JSON-encoded message to the bus. + """ + ... + +class Cancellable(Object): + def cancel(self) -> None: + """ + Set cancellable to cancelled. + """ + ... + def connect(self, callback: Callable[..., Any]) -> int: + """ + Register notification callback. + """ + ... + def disconnect(self, handler_id: int) -> None: + """ + Unregister notification callback. + """ + ... + @classmethod + def get_current(cls) -> "Cancellable": + """ + Get the top cancellable from the stack. + """ + ... + def get_fd(self) -> int: + """ + Get file descriptor for integrating with an event loop. + """ + ... + def is_cancelled(self) -> bool: + """ + Query whether cancellable has been cancelled. + """ + ... + def pop_current(self) -> None: + """ + Pop cancellable off the cancellable stack. + """ + ... + def push_current(self) -> None: + """ + Push cancellable onto the cancellable stack. + """ + ... + def raise_if_cancelled(self) -> None: + """ + Raise an exception if cancelled. + """ + ... + def release_fd(self) -> None: + """ + Release a resource previously allocated by get_fd(). + """ + ... + +class Child(Object): + @property + def argv(self) -> List[str]: + """ + Argument vector. + """ + ... + @property + def envp(self) -> Dict[str, str]: + """ + Environment vector. + """ + ... + @property + def identifier(self) -> str: + """ + Application identifier. + """ + ... + @property + def origin(self) -> str: + """ + Origin. + """ + ... + @property + def parent_pid(self) -> int: + """ + Parent Process ID. + """ + ... + @property + def path(self) -> str: + """ + Path of executable. + """ + ... + @property + def pid(self) -> int: + """ + Process ID. + """ + ... + +class Crash(Object): + @property + def parameters(self) -> Dict[str, Any]: + """ + Parameters. + """ + ... + @property + def pid(self) -> int: + """ + Process ID. + """ + ... + @property + def process_name(self) -> str: + """ + Process name. + """ + ... + @property + def report(self) -> str: + """ + Human-readable crash report. + """ + ... + @property + def summary(self) -> str: + """ + Human-readable crash summary. + """ + ... + +class Device(Object): + @property + def id(self) -> Optional[str]: + """ + Device ID. + """ + ... + @property + def name(self) -> Optional[str]: + """ + Human-readable device name. + """ + ... + @property + def icon(self) -> Optional[Any]: + """ + Icon. + """ + ... + @property + def type(self) -> Optional[str]: + """ + Device type. One of: local, remote, usb. + """ + ... + @property + def bus(self) -> Optional[Bus]: + """ + Message bus. + """ + ... + def attach(self, pid: int, realm: Optional[str] = None, persist_timeout: Optional[int] = None) -> "Session": + """ + Attach to a PID. + """ + ... + def disable_spawn_gating(self) -> None: + """ + Disable spawn gating. + """ + ... + def enable_spawn_gating(self) -> None: + """ + Enable spawn gating. + """ + ... + def enumerate_applications( + self, identifiers: Optional[Sequence[str]] = None, scope: Optional[str] = None + ) -> List[Application]: + """ + Enumerate applications. + """ + ... + def enumerate_pending_children(self) -> List[Child]: + """ + Enumerate pending children. + """ + ... + def enumerate_pending_spawn(self) -> List["Spawn"]: + """ + Enumerate pending spawn. + """ + ... + def enumerate_processes(self, pids: Optional[Sequence[int]] = None, scope: Optional[str] = None) -> List[Process]: + """ + Enumerate processes. + """ + ... + def get_frontmost_application(self, scope: Optional[str] = None) -> Optional[Application]: + """ + Get details about the frontmost application. + """ + ... + def inject_library_blob(self, pid: int, blob_buffer: bytes, entrypoint: str, data: str) -> int: + """ + Inject a library blob to a PID. + """ + ... + def inject_library_file(self, pid: int, path: str, entrypoint: str, data: str) -> int: + """ + Inject a library file to a PID. + """ + ... + def input(self, pid: int, data: bytes) -> None: + """ + Input data on stdin of a spawned process. + """ + ... + def is_lost(self) -> bool: + """ + Query whether the device has been lost. + """ + ... + def kill(self, pid: int) -> None: + """ + Kill a PID. + """ + ... + def open_channel(self, address: str) -> "IOStream": + """ + Open a device-specific communication channel. + """ + ... + def query_system_parameters(self) -> Dict[str, Any]: + """ + Returns a dictionary of information about the host system. + """ + ... + def resume(self, pid: int) -> None: + """ + Resume a process from the attachable state. + """ + ... + def spawn( + self, + program: str, + argv: Union[None, List[Union[str, bytes]], Tuple[Union[str, bytes]]] = None, + envp: Optional[Dict[str, str]] = None, + env: Optional[Dict[str, str]] = None, + cwd: Optional[str] = None, + stdio: Optional[str] = None, + **kwargs: Any, + ) -> int: + """ + Spawn a process into an attachable state. + """ + ... + +class DeviceManager(Object): + def add_remote_device( + self, + address: str, + certificate: Optional[str] = None, + origin: Optional[str] = None, + token: Optional[str] = None, + keepalive_interval: Optional[int] = None, + ) -> Device: + """ + Add a remote device. + """ + ... + def close(self) -> None: + """ + Close the device manager. + """ + ... + def enumerate_devices(self) -> List[Device]: + """ + Enumerate devices. + """ + ... + def get_device_matching(self, predicate: Callable[[Device], bool], timeout: int) -> Device: + """ + Get device matching predicate. + """ + ... + def remove_remote_device(self, address: str) -> None: + """ + Remove a remote device. + """ + ... + +class EndpointParameters(Object): ... + +class FileMonitor(Object): + def disable(self) -> None: + """ + Disable the file monitor. + """ + ... + def enable(self) -> None: + """ + Enable the file monitor. + """ + ... + +class IOStream(Object): + def close(self) -> None: + """ + Close the stream. + """ + ... + def is_closed(self) -> bool: + """ + Query whether the stream is closed. + """ + ... + def read(self, size: int) -> bytes: + """ + Read up to the specified number of bytes from the stream. + """ + ... + def read_all(self, size: int) -> bytes: + """ + Read exactly the specified number of bytes from the stream. + """ + ... + def write(self, data: bytes) -> int: + """ + Write as much as possible of the provided data to the stream. + """ + ... + def write_all(self, data: bytes) -> None: + """ + Write all of the provided data to the stream. + """ + ... + +class PortalMembership(Object): + def terminate(self) -> None: + """ + Terminate the membership. + """ + ... + +class PortalService(Object): + @property + def device(self) -> Device: + """ + Device for in-process control. + """ + ... + def broadcast(self, message: str, data: Optional[Union[str, bytes]] = None) -> None: + """ + Broadcast a message to all control channels. + """ + ... + def enumerate_tags(self, connection_id: int) -> List[str]: + """ + Enumerate tags of a specific connection. + """ + ... + def kick(self, connection_id: int) -> None: + """ + Kick out a specific connection. + """ + ... + def narrowcast(self, tag: str, message: str, data: Optional[Union[str, bytes]] = None) -> None: + """ + Post a message to control channels with a specific tag. + """ + ... + def post(self, connection_id: int, message: str, data: Optional[Union[str, bytes]] = None) -> None: + """ + Post a message to a specific control channel. + """ + ... + def start(self) -> None: + """ + Start listening for incoming connections. + """ + ... + def stop(self) -> None: + """ + Stop listening for incoming connections, and kick any connected clients. + """ + ... + def tag(self, connection_id: int, tag: str) -> None: + """ + Tag a specific control channel. + """ + ... + def untag(self, connection_id: int, tag: str) -> None: + """ + Untag a specific control channel. + """ + ... + +class Process(Object): + @property + def pid(self) -> int: + """ + Process ID. + """ + ... + @property + def name(self) -> str: + """ + Human-readable process name. + """ + ... + @property + def parameters(self) -> Dict[str, Any]: + """ + Parameters. + """ + ... + +class Relay(Object): + def __init__(self, address: str, username: str, password: str, kind: str) -> None: ... + @property + def address(self) -> str: + """ + Network address or address:port of the TURN server. + """ + ... + @property + def kind(self) -> str: + """ + Relay kind. One of: turn-udp, turn-tcp, turn-tls. + """ + ... + @property + def password(self) -> str: + """ + The TURN password to use for the allocate request. + """ + ... + @property + def username(self) -> str: + """ + The TURN username to use for the allocate request. + """ + ... + +class Script(Object): + def eternalize(self) -> None: + """ + Eternalize the script. + """ + ... + def is_destroyed(self) -> bool: + """ + Query whether the script has been destroyed. + """ + ... + def load(self) -> None: + """ + Load the script. + """ + ... + def post(self, message: str, data: Optional[Union[str, bytes]] = None) -> None: + """ + Post a JSON-encoded message to the script. + """ + ... + def unload(self) -> None: + """ + Unload the script. + """ + ... + def enable_debugger(self, port: Optional[int]) -> None: + """ + Enable the Node.js compatible script debugger + """ + ... + def disable_debugger(self) -> None: + """ + Disable the Node.js compatible script debugger + """ + ... + +class Session(Object): + @property + def pid(self) -> int: + """ + Process ID. + """ + ... + def compile_script(self, source: str, name: Optional[str] = None, runtime: Optional[str] = None) -> bytes: + """ + Compile script source code to bytecode. + """ + ... + def create_script(self, source: str, name: Optional[str] = None, runtime: Optional[str] = None) -> Script: + """ + Create a new script. + """ + ... + def create_script_from_bytes( + self, data: bytes, name: Optional[str] = None, runtime: Optional[str] = None + ) -> Script: + """ + Create a new script from bytecode. + """ + ... + def snapshot_script(self, embed_script: str, warmup_script: Optional[str], runtime: Optional[str] = None) -> bytes: + """ + Evaluate script and snapshot the resulting VM state + """ + ... + def detach(self) -> None: + """ + Detach session from the process. + """ + ... + def disable_child_gating(self) -> None: + """ + Disable child gating. + """ + ... + def enable_child_gating(self) -> None: + """ + Enable child gating. + """ + ... + def is_detached(self) -> bool: + """ + Query whether the session is detached. + """ + ... + def join_portal( + self, address: str, certificate: Optional[str] = None, token: Optional[str] = None, acl: Optional[Any] = None + ) -> PortalMembership: + """ + Join a portal. + """ + ... + def resume(self) -> None: + """ + Resume session after network error. + """ + ... + def setup_peer_connection( + self, stun_server: Optional[str] = None, relays: Optional[Sequence[Relay]] = None + ) -> None: + """ + Set up a peer connection with the target process. + """ + ... + +class Spawn(Object): + @property + def identifier(self) -> str: + """ + Application identifier. + """ + ... + @property + def pid(self) -> int: + """ + Process ID. + """ + ... + +class Compiler(Object): + def build( + self, + entrypoint: str, + project_root: Optional[str] = None, + source_maps: Optional[str] = None, + compression: Optional[str] = None, + ) -> str: + """ + Build an agent. + """ + ... + def watch( + self, + entrypoint: str, + project_root: Optional[str] = None, + source_maps: Optional[str] = None, + compression: Optional[str] = None, + ) -> None: + """ + Continuously build an agent. + """ + ... + +__version__: str diff --git a/_frida/py.typed b/_frida/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/examples/bytecode.py b/examples/bytecode.py index 30712ed..d7abdfa 100644 --- a/examples/bytecode.py +++ b/examples/bytecode.py @@ -1,17 +1,16 @@ -# -*- coding: utf-8 -*- -from __future__ import print_function - 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 08fb670..fe8dbb4 100644 --- a/examples/channels.py +++ b/examples/channels.py @@ -1,8 +1,5 @@ -# -*- coding: utf-8 -*- -from __future__ import unicode_literals, print_function 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 d2ad5b7..0e90ac5 100644 --- a/examples/child_gating.py +++ b/examples/child_gating.py @@ -1,13 +1,11 @@ -# -*- coding: utf-8 -*- -from __future__ import print_function - import threading -import frida from frida_tools.application import Reactor +import frida -class Application(object): + +class Application: def __init__(self): self._stop_requested = threading.Event() self._reactor = Reactor(run_until_return=lambda reactor: self._stop_requested.wait()) @@ -29,8 +27,8 @@ class Application(object): "BADGER": "badger-badger-badger", "SNAKE": "mushroom-mushroom", } - print("✔ spawn(argv={})".format(argv)) - pid = self._device.spawn(argv, env=env, stdio='pipe') + print(f"✔ spawn(argv={argv})") + pid = self._device.spawn(argv, env=env, stdio="pipe") self._instrument(pid) def _stop_if_idle(self): @@ -38,13 +36,14 @@ class Application(object): self._stop_requested.set() def _instrument(self, pid): - print("✔ attach(pid={})".format(pid)) + print(f"✔ attach(pid={pid})") session = self._device.attach(pid) session.on("detached", lambda reason: self._reactor.schedule(lambda: self._on_detached(pid, session, reason))) 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({ @@ -53,31 +52,32 @@ 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() - print("✔ resume(pid={})".format(pid)) + print(f"✔ resume(pid={pid})") self._device.resume(pid) self._sessions.add(session) def _on_child_added(self, child): - print("⚡ child_added: {}".format(child)) + print(f"⚡ child_added: {child}") self._instrument(child.pid) def _on_child_removed(self, child): - print("⚡ child_removed: {}".format(child)) + print(f"⚡ child_removed: {child}") def _on_output(self, pid, fd, data): - print("⚡ output: pid={}, fd={}, data={}".format(pid, fd, repr(data))) + print(f"⚡ output: pid={pid}, fd={fd}, data={repr(data)}") def _on_detached(self, pid, session, reason): - print("⚡ detached: pid={}, reason='{}'".format(pid, reason)) + print(f"⚡ detached: pid={pid}, reason='{reason}'") self._sessions.remove(session) self._reactor.schedule(self._stop_if_idle, delay=0.5) def _on_message(self, pid, message): - print("⚡ message: pid={}, payload={}".format(pid, message["payload"])) + print(f"⚡ message: pid={pid}, payload={message['payload']}") app = Application() 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 9db9913..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,13 +59,11 @@ 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, e: + except Exception as e: if session is not None: try: session.detach() @@ -93,8 +93,8 @@ class Capture(NSObject): pool = NSAutoreleasePool.alloc().init() try: script.post(message) - except Exception, e: - print "Failed to post to script:", e + except Exception as e: + print("Failed to post to script:", e) del pool def _attachDidCompleteWithSession_script_error_(self, session, script, error): @@ -113,22 +113,22 @@ 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: if not self.calls._handleStanza_(stanza): - print "Woot! Got stanza: %s from=%s" % (stanza['name'], stanza['from']) + print(f"Woot! Got stanza: {stanza['name']} from={stanza['from']}") else: - print "Unhandled message:", message + print("Unhandled message:", message) def _onSessionDetached(self): AppHelper.callAfter(self._sessionDidDetach) @@ -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) @@ -355,9 +350,10 @@ class FunctionCall(NSObject): self = self.init() self.func = func self.args = args - self.summary = "%s(%s)" % (func.name, ", ".join(args)) + 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 d5f56bf..736f672 100644 --- a/examples/crash_reporting.py +++ b/examples/crash_reporting.py @@ -1,6 +1,3 @@ -# -*- coding: utf-8 -*- -from __future__ import print_function - import sys import frida @@ -10,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 dac552d..638c2a8 100644 --- a/examples/detached.py +++ b/examples/detached.py @@ -1,6 +1,3 @@ -# -*- coding: utf-8 -*- -from __future__ import print_function - import sys import frida @@ -9,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 3572a8e..075df30 100644 --- a/examples/enumerate_applications.py +++ b/examples/enumerate_applications.py @@ -1,26 +1,24 @@ -# -*- coding: utf-8 -*- -from __future__ import print_function -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']] - print("Application(identifier=\"{}\", name=\"{}\", pid={}, parameters={})".format( - app.identifier, - app.name, - app.pid, - highlight(pformat(params), PythonLexer(), Terminal256Formatter()).rstrip())) + 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})') diff --git a/examples/enumerate_processes.py b/examples/enumerate_processes.py index 6c2eda9..a78d8e0 100644 --- a/examples/enumerate_processes.py +++ b/examples/enumerate_processes.py @@ -1,22 +1,25 @@ -# -*- coding: utf-8 -*- -from __future__ import print_function -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("Process(pid={}, name=\"{}\", parameters={})".format(proc.pid, proc.name, 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 0b14841..708fb11 100644 --- a/examples/get_frontmost_application.py +++ b/examples/get_frontmost_application.py @@ -1,24 +1,25 @@ -# -*- coding: utf-8 -*- -from __future__ import print_function -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']] - print("{}:".format(app.identifier), highlight(pformat(params), PythonLexer(), Terminal256Formatter())) + 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 b567292..5d3c817 100644 --- a/examples/inject_library/inject_blob.py +++ b/examples/inject_library/inject_blob.py @@ -6,8 +6,6 @@ # $ python inject_blob.py Twitter example.dylib # -from __future__ import unicode_literals, print_function - import sys import frida @@ -16,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 32fd008..cd104ed 100644 --- a/examples/inject_library/inject_file.py +++ b/examples/inject_library/inject_file.py @@ -6,8 +6,6 @@ # $ python inject_file.py Twitter ~/.Trash/example.dylib # -from __future__ import unicode_literals, print_function - import sys import frida @@ -16,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 9608536..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,54 +43,44 @@ 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 = "{} > ".format(channel) - self._bus.post({ - 'type': 'join', - 'channel': channel - }) + self._prompt = f"{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(["{} (connected from {})".format(m['nick'], m['address']) for m in message['members']])) - for item in message['history']: - self._print("<{}> {}".format(item['sender'], item['text'])) - elif mtype == 'join': - user = message['user'] - self._print("👋 {} ({}) joined {}".format(user['nick'], user['address'], message['channel'])) - elif mtype == 'part': - user = message['user'] - self._print("🚪 {} ({}) left {}".format(user['nick'], user['address'], message['channel'])) - elif mtype == 'chat': - self._print("<{}> {}".format(message['sender'], message['text'])) - elif mtype == 'announce': - self._print("📣 <{}> {}".format(message['sender'], message['text'])) + 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"] + self._print(f"👋 {user['nick']} ({user['address']}) joined {message['channel']}") + elif mtype == "part": + user = message["user"] + self._print(f"🚪 {user['nick']} ({user['address']}) left {message['channel']}") + elif mtype == "chat": + self._print(f"<{message['sender']}> {message['text']}") + 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 928d80e..a99066b 100644 --- a/examples/portal_server.py +++ b/examples/portal_server.py @@ -1,11 +1,11 @@ -# -*- coding: utf-8 -*- -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 @@ -14,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 @@ -34,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) @@ -75,18 +81,18 @@ 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") provided = hashlib.sha1(secret).digest() - expected = hashlib.sha1("knock-knock".encode('utf-8')).digest() + expected = hashlib.sha1(b"knock-knock").digest() if not hmac.compare_digest(provided, expected): raise ValueError("get outta here") return { - 'nick': nick, + "nick": nick, } def _on_node_connected(self, connection_id, remote_address): @@ -119,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) @@ -184,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: @@ -205,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: @@ -227,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) @@ -251,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 01dfa2a..f5f38d9 100644 --- a/examples/query_system_parameters.py +++ b/examples/query_system_parameters.py @@ -1,11 +1,13 @@ -# -*- coding: utf-8 -*- -from __future__ import print_function -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 fc031b5..5684a3b 100644 --- a/examples/rpc.py +++ b/examples/rpc.py @@ -1,11 +1,8 @@ -# -*- coding: utf-8 -*- -from __future__ import print_function - import frida - session = frida.attach("Twitter") -script = session.create_script("""\ +script = session.create_script( + """\ rpc.exports = { hello: function () { return 'Hello'; @@ -14,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 e0199e6..d749d11 100644 --- a/examples/session_persist_timeout.py +++ b/examples/session_persist_timeout.py @@ -1,9 +1,9 @@ -# -*- coding: utf-8 -*- -import frida from frida_tools.application import Reactor +import frida -class Application(object): + +class Application: def __init__(self): self._reactor = Reactor(run_until_return=self._process_input) @@ -20,9 +20,10 @@ class Application(object): 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'), { @@ -47,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): @@ -69,10 +71,10 @@ function puts(s) { print("Unknown command") def _on_detached(self, reason, crash): - print("⚡ detached: reason={}, crash={}".format(reason, crash)) + print(f"⚡ detached: reason={reason}, crash={crash}") def _on_message(self, message, data): - print("⚡ message: {}".format(message)) + print(f"⚡ message: {message}") app = Application() 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 8bc1071..a133dfb 100644 --- a/frida/__init__.py +++ b/frida/__init__.py @@ -1,31 +1,22 @@ -# -*- coding: utf-8 -*- -from __future__ import unicode_literals, print_function - -import threading +from typing import Any, Callable, Dict, List, Optional, Tuple, Union try: import _frida except Exception as ex: - import sys print("") print("***") if str(ex).startswith("No module named "): print("Frida native extension not found") print("Please check your PYTHONPATH.") else: - print("Failed to load the Frida native extension: %s" % ex) - if sys.version_info[0] == 2: - current_python_version = "%d.%d" % sys.version_info[:2] - else: - current_python_version = "%d.x" % sys.version_info[0] - print("Please ensure that the extension was compiled for Python " + current_python_version + ".") + print(f"Failed to load the Frida native extension: {ex}") + print("Please ensure that the extension was compiled correctly") print("***") print("") raise ex from . import core - -__version__ = _frida.__version__ +__version__: str = _frida.__version__ get_device_manager = core.get_device_manager Relay = _frida.Relay @@ -51,58 +42,131 @@ TransportError = _frida.TransportError OperationCancelledError = _frida.OperationCancelledError -def query_system_parameters(**kwargs): - return get_local_device().query_system_parameters(**kwargs) +def query_system_parameters() -> Dict[str, Any]: + """ + Returns a dictionary of information about the host system + """ + + return get_local_device().query_system_parameters() -def spawn(*args, **kwargs): - return get_local_device().spawn(*args, **kwargs) +def spawn( + program: Union[str, List[Union[str, bytes]], Tuple[Union[str, bytes]]], + argv: Union[None, List[Union[str, bytes]], Tuple[Union[str, bytes]]] = None, + envp: Optional[Dict[str, str]] = None, + env: Optional[Dict[str, str]] = None, + cwd: Optional[str] = None, + stdio: Optional[str] = None, + **kwargs: Any, +) -> int: + """ + Spawn a process into an attachable state + """ + + return get_local_device().spawn(program=program, argv=argv, envp=envp, env=env, cwd=cwd, stdio=stdio, **kwargs) -def resume(target, **kwargs): - get_local_device().resume(target, **kwargs) +def resume(target: core.ProcessTarget) -> None: + """ + Resume a process from the attachable state + :param target: the PID or name of the process + """ + + get_local_device().resume(target) -def kill(target, **kwargs): - get_local_device().kill(target, **kwargs) +def kill(target: core.ProcessTarget) -> None: + """ + Kill a process + :param target: the PID or name of the process + """ + + get_local_device().kill(target) -def attach(target, *args, **kwargs): - return get_local_device().attach(target, *args, **kwargs) +def attach( + target: core.ProcessTarget, realm: Optional[str] = None, persist_timeout: Optional[int] = None +) -> core.Session: + """ + Attach to a process + :param target: the PID or name of the process + """ + + return get_local_device().attach(target, realm=realm, persist_timeout=persist_timeout) -def inject_library_file(target, path, entrypoint, data, **kwargs): - return get_local_device().inject_library_file(target, path, entrypoint, data, **kwargs) +def inject_library_file(target: core.ProcessTarget, path: str, entrypoint: str, data: str) -> int: + """ + Inject a library file to a process. + :param target: the PID or name of the process + """ + + return get_local_device().inject_library_file(target, path, entrypoint, data) -def inject_library_blob(target, blob, entrypoint, data, **kwargs): - return get_local_device().inject_library_blob(target, blob, entrypoint, data, **kwargs) +def inject_library_blob(target: core.ProcessTarget, blob: bytes, entrypoint: str, data: str) -> int: + """ + Inject a library blob to a process + :param target: the PID or name of the process + """ + + return get_local_device().inject_library_blob(target, blob, entrypoint, data) -def get_local_device(**kwargs): - return get_device_matching(lambda d: d.type == 'local', timeout=0, **kwargs) +def get_local_device() -> core.Device: + """ + Get the local device + """ + + return get_device_manager().get_local_device() -def get_remote_device(**kwargs): - return get_device_matching(lambda d: d.type == 'remote', timeout=0, **kwargs) +def get_remote_device() -> core.Device: + """ + Get the first remote device in the devices list + """ + + return get_device_manager().get_remote_device() -def get_usb_device(timeout=0, **kwargs): - return get_device_matching(lambda d: d.type == 'usb', timeout, **kwargs) +def get_usb_device(timeout: int = 0) -> core.Device: + """ + Get the first device connected over USB in the devices list + """ + + return get_device_manager().get_usb_device(timeout) -def get_device(id, timeout=0, **kwargs): - return get_device_manager().get_device(id, timeout, **kwargs) +def get_device(id: Optional[str], timeout: int = 0) -> core.Device: + """ + Get a device by its id + """ + + return get_device_manager().get_device(id, timeout) -def get_device_matching(predicate, timeout=0, **kwargs): - return get_device_manager().get_device_matching(predicate, timeout, **kwargs) +def get_device_matching(predicate: Callable[[core.Device], bool], timeout: int = 0) -> core.Device: + """ + Get device matching predicate. + :param predicate: a function to filter the devices + :param timeout: operation timeout in seconds + """ + + return get_device_manager().get_device_matching(predicate, timeout) -def enumerate_devices(**kwargs): - return get_device_manager().enumerate_devices(**kwargs) +def enumerate_devices() -> List[core.Device]: + """ + Enumerate all the devices from the device manager + """ + + return get_device_manager().enumerate_devices() @core.cancellable -def shutdown(): +def shutdown() -> None: + """ + Shutdown the main device manager + """ + get_device_manager()._impl.close() diff --git a/frida/core.py b/frida/core.py index dd95d2a..bf8d9d4 100644 --- a/frida/core.py +++ b/frida/core.py @@ -1,32 +1,45 @@ -# -*- coding: utf-8 -*- -from __future__ import unicode_literals, print_function - import fnmatch -from functools import wraps +import functools import json -import numbers import sys import threading import traceback +from types import TracebackType +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Type, TypeVar, Union import _frida - _device_manager = None + _Cancellable = _frida.Cancellable +ProcessTarget = Union[int, str] + + +def get_device_manager() -> "DeviceManager": + """ + Get or create a singleton DeviceManager that let you manage all the devices + """ -def get_device_manager(): global _device_manager if _device_manager is None: _device_manager = DeviceManager(_frida.DeviceManager()) return _device_manager -def cancellable(f): - @wraps(f) - def wrapper(*args, **kwargs): - cancellable = kwargs.pop('cancellable', None) +def _filter_missing_kwargs(d: Dict[str, Any]) -> None: + for key in list(d.keys()): + if d[key] is None: + d.pop(key) + + +R = TypeVar("R") + + +def cancellable(f: Callable[..., R]) -> Callable[..., R]: + @functools.wraps(f) + def wrapper(*args: Any, **kwargs: Any) -> R: + cancellable = kwargs.pop("cancellable", None) if cancellable is not None: with cancellable: return f(*args, **kwargs) @@ -36,352 +49,250 @@ def cancellable(f): return wrapper -class DeviceManager(object): - def __init__(self, impl): +class IOStream: + """ + Frida's own implementation of an input/output stream + """ + + def __init__(self, impl: _frida.IOStream) -> None: self._impl = impl - def __repr__(self): - return repr(self._impl) - - def get_local_device(self, **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) - - def get_usb_device(self, timeout=0, **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) - - @cancellable - def get_device_matching(self, predicate, timeout=0): - if timeout < 0: - raw_timeout = -1 - elif timeout == 0: - raw_timeout = 0 - else: - raw_timeout = int(timeout * 1000.0) - return Device(self._impl.get_device_matching(lambda d: predicate(Device(d)), raw_timeout)) - - @cancellable - def enumerate_devices(self): - return [Device(device) for device in self._impl.enumerate_devices()] - - @cancellable - def add_remote_device(self, *args, **kwargs): - return Device(self._impl.add_remote_device(*args, **kwargs)) - - @cancellable - def remove_remote_device(self, *args, **kwargs): - self._impl.remove_remote_device(*args, **kwargs) - - def on(self, signal, callback): - self._impl.on(signal, callback) - - def off(self, signal, callback): - self._impl.off(signal, callback) - - -class Device(object): - def __init__(self, device): - self.id = device.id - self.name = device.name - self.icon = device.icon - self.type = device.type - self.bus = Bus(device.bus) - - self._impl = device - - def __repr__(self): + def __repr__(self) -> str: return repr(self._impl) @property - def is_lost(self): - return self._impl.is_lost() + def is_closed(self) -> bool: + """ + Query whether the stream is closed + """ + + return self._impl.is_closed() @cancellable - def query_system_parameters(self): - return self._impl.query_system_parameters() + def close(self) -> None: + """ + Close the stream. + """ + + self._impl.close() @cancellable - def get_frontmost_application(self, *args, **kwargs): - return self._impl.get_frontmost_application(*args, **kwargs) + def read(self, count: int) -> bytes: + """ + Read up to the specified number of bytes from the stream + """ + + return self._impl.read(count) @cancellable - def enumerate_applications(self, *args, **kwargs): - return self._impl.enumerate_applications(*args, **kwargs) + def read_all(self, count: int) -> bytes: + """ + Read exactly the specified number of bytes from the stream + """ + + return self._impl.read_all(count) @cancellable - def enumerate_processes(self, *args, **kwargs): - return self._impl.enumerate_processes(*args, **kwargs) + def write(self, data: bytes) -> int: + """ + Write as much as possible of the provided data to the stream + """ + + return self._impl.write(data) @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)] - if len(matching) == 1: - return matching[0] - elif len(matching) > 1: - raise _frida.ProcessNotFoundError("ambiguous name; it matches: %s" % ", ".join(["%s (pid: %d)" % (process.name, process.pid) for process in matching])) - else: - raise _frida.ProcessNotFoundError("unable to find process with name '%s'" % process_name) + def write_all(self, data: bytes) -> None: + """ + Write all of the provided data to the stream + """ - @cancellable - def enable_spawn_gating(self): - return self._impl.enable_spawn_gating() - - @cancellable - def disable_spawn_gating(self): - return self._impl.disable_spawn_gating() - - @cancellable - def enumerate_pending_spawn(self): - return self._impl.enumerate_pending_spawn() - - @cancellable - def enumerate_pending_children(self): - return self._impl.enumerate_pending_children() - - @cancellable - def spawn(self, program, argv=None, envp=None, env=None, cwd=None, stdio=None, **kwargs): - if not isinstance(program, string_types): - argv = program - program = argv[0] - if len(argv) == 1: - argv = None - - aux_options = kwargs - - return self._impl.spawn(program, argv, envp, env, cwd, stdio, aux_options) - - @cancellable - def input(self, target, data): - self._impl.input(self._pid_of(target), data) - - @cancellable - def resume(self, target): - self._impl.resume(self._pid_of(target)) - - @cancellable - def kill(self, target): - self._impl.kill(self._pid_of(target)) - - @cancellable - def attach(self, target, *args, **kwargs): - return Session(self._impl.attach(self._pid_of(target), *args, **kwargs)) - - @cancellable - def inject_library_file(self, target, path, entrypoint, data): - return self._impl.inject_library_file(self._pid_of(target), path, entrypoint, data) - - @cancellable - def inject_library_blob(self, target, blob, entrypoint, data): - return self._impl.inject_library_blob(self._pid_of(target), blob, entrypoint, data) - - @cancellable - def open_channel(self, address): - return IOStream(self._impl.open_channel(address)) - - @cancellable - def get_bus(self): - return Bus(self._impl.get_bus()) - - def on(self, signal, callback): - self._impl.on(signal, callback) - - def off(self, signal, callback): - self._impl.off(signal, callback) - - def _pid_of(self, target): - if isinstance(target, numbers.Number): - return target - else: - return self.get_process(target).pid + self._impl.write_all(data) -class Bus(object): - def __init__(self, impl): - self._impl = impl - self._on_message_callbacks = [] - - impl.on('message', self._on_message) - - @cancellable - def attach(self): - self._impl.attach() - - def post(self, message, **kwargs): - raw_message = json.dumps(message) - self._impl.post(raw_message, **kwargs) - - def on(self, signal, callback): - if signal == 'message': - self._on_message_callbacks.append(callback) - else: - self._impl.on(signal, callback) - - def off(self, signal, callback): - if signal == 'message': - self._on_message_callbacks.remove(callback) - else: - self._impl.off(signal, callback) - - def _on_message(self, raw_message, data): - message = json.loads(raw_message) - - for callback in self._on_message_callbacks[:]: - try: - callback(message, data) - except: - traceback.print_exc() - - -class Session(object): - def __init__(self, impl): +class PortalMembership: + def __init__(self, impl: _frida.PortalMembership) -> None: self._impl = impl - def __repr__(self): - return repr(self._impl) - - @property - def is_detached(self): - return self._impl.is_detached() - @cancellable - def detach(self): - self._impl.detach() + def terminate(self) -> None: + """ + Terminate the membership + """ - @cancellable - def resume(self): - self._impl.resume() - - @cancellable - def enable_child_gating(self): - self._impl.enable_child_gating() - - @cancellable - def disable_child_gating(self): - self._impl.disable_child_gating() - - @cancellable - def create_script(self, *args, **kwargs): - return Script(self._impl.create_script(*args, **kwargs)) - - @cancellable - def create_script_from_bytes(self, *args, **kwargs): - return Script(self._impl.create_script_from_bytes(*args, **kwargs)) - - @cancellable - def compile_script(self, *args, **kwargs): - return self._impl.compile_script(*args, **kwargs) - - @cancellable - def snapshot_script(self, *args, **kwargs): - return self._impl.snapshot_script(*args, **kwargs) - - @cancellable - def setup_peer_connection(self, *args, **kwargs): - self._impl.setup_peer_connection(*args, **kwargs) - - @cancellable - def join_portal(self, *args, **kwargs): - return PortalMembership(self._impl.join_portal(*args, **kwargs)) - - def on(self, signal, callback): - self._impl.on(signal, callback) - - def off(self, signal, callback): - self._impl.off(signal, callback) + self._impl.terminate() -class Script(object): - def __init__(self, impl): +class ScriptExports: + """ + Proxy object that expose all the RPC exports of a script as attributes on this class + + A method named exampleMethod in a script will be called with instance.example_method on this object + """ + + def __init__(self, script: "Script") -> None: + self._script = script + + def __getattr__(self, name: str) -> Any: + script = self._script + js_name = _to_camel_case(name) + + def method(*args: Any, **kwargs: Any) -> Any: + return script._rpc_request("call", js_name, args, **kwargs) + + return method + + def __dir__(self) -> List[str]: + return self._script.list_exports() + + +class Script: + def __init__(self, impl: _frida.Script) -> None: self.exports = ScriptExports(self) self._impl = impl - self._on_message_callbacks = [] - self._log_handler = self.default_log_handler + self._on_message_callbacks: List[Callable[..., Any]] = [] + self._log_handler: Callable[[str, str], None] = self.default_log_handler - self._pending = {} + self._pending: Dict[int, Callable[..., Any]] = {} 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): + def __repr__(self) -> str: return repr(self._impl) @property - def is_destroyed(self): + def is_destroyed(self) -> bool: + """ + Query whether the script has been destroyed + """ + return self._impl.is_destroyed() @cancellable - def load(self): + def load(self) -> None: + """ + Load the script. + """ + self._impl.load() @cancellable - def unload(self): + def unload(self) -> None: + """ + Unload the script + """ + self._impl.unload() @cancellable - def eternalize(self): + def eternalize(self) -> None: + """ + Eternalize the script + """ + self._impl.eternalize() - def post(self, message, **kwargs): + def post(self, message: Any, data: Optional[str] = None) -> None: + """ + Post a JSON-encoded message to the script + """ + raw_message = json.dumps(message) + kwargs = {"data": data} + _filter_missing_kwargs(kwargs) self._impl.post(raw_message, **kwargs) @cancellable - def enable_debugger(self, *args, **kwargs): - self._impl.enable_debugger(*args, **kwargs) + def enable_debugger(self, port: Optional[int] = None) -> None: + """ + Enable the Node.js compatible script debugger + """ + + kwargs = {"port": port} + _filter_missing_kwargs(kwargs) + self._impl.enable_debugger(**kwargs) @cancellable - def disable_debugger(self): + def disable_debugger(self) -> None: + """ + Disable the Node.js compatible script debugger + """ + self._impl.disable_debugger() - def on(self, signal, callback): - if signal == 'message': + def on(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Add a signal handler + """ + + if signal == "message": self._on_message_callbacks.append(callback) else: self._impl.on(signal, callback) - def off(self, signal, callback): - if signal == 'message': + def off(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Remove a signal handler + """ + + if signal == "message": self._on_message_callbacks.remove(callback) else: self._impl.off(signal, callback) - def get_log_handler(self): + def get_log_handler(self) -> Callable[[str, str], None]: + """ + Get the method that handles the script logs + """ + return self._log_handler - def set_log_handler(self, handler): + def set_log_handler(self, handler: Callable[[str, str], None]) -> None: + """ + Set the method that handles the script logs + :param handler: a callable that accepts two parameters: + 1. the log level name + 2. the log message + """ + self._log_handler = handler - def default_log_handler(self, level, text): - if level == 'info': + def default_log_handler(self, level: str, text: str) -> None: + """ + The default implementation of the log handler, prints the message to stdout + or stderr, depending on the level + """ + + if level == "info": print(text, file=sys.stdout) else: print(text, file=sys.stderr) - def list_exports(self): - return self._rpc_request('list') + def list_exports(self) -> List[str]: + """ + List all the exported attributes from the script's rpc + """ + + result = self._rpc_request("list") + assert isinstance(result, list) + return result @cancellable - def _rpc_request(self, *args): + def _rpc_request(self, *args: Any) -> Any: result = [False, None, None] - def on_complete(value, error): + def on_complete(value: Any, error: Union[None, RPCException | _frida.InvalidOperationError]) -> None: with self._cond: result[0] = True result[1] = value result[2] = error self._cond.notify_all() - def on_cancelled(): + def on_cancelled() -> None: self._pending.pop(request_id, None) on_complete(None, None) @@ -391,7 +302,7 @@ class Script(object): 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) @@ -413,22 +324,22 @@ class Script(object): return result[1] - def _on_rpc_message(self, request_id, operation, params, data): - if operation in ('ok', 'error'): + def _on_rpc_message(self, request_id: int, operation: str, params, data) -> None: + 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]) callback(value, error) - def _on_destroyed(self): + def _on_destroyed(self) -> None: while True: next_pending = None @@ -440,18 +351,18 @@ class Script(object): 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): + def _on_message(self, raw_message: str, data: Any) -> None: 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:] @@ -464,68 +375,589 @@ class Script(object): traceback.print_exc() -class RPCException(Exception): - def __str__(self): - return self.args[2] if len(self.args) >= 3 else self.args[0] - - -class ScriptExports(object): - def __init__(self, script): - self._script = script - - 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 method - - def __dir__(self): - return self._script.list_exports() - - -class PortalMembership(object): - def __init__(self, impl): +class Session: + def __init__(self, impl: _frida.Session) -> None: self._impl = impl + def __repr__(self) -> str: + return repr(self._impl) + + @property + def is_detached(self) -> bool: + """ + Query whether the session is detached + """ + + return self._impl.is_detached() + @cancellable - def terminate(self): - self._impl.terminate() + def detach(self) -> None: + """ + Detach session from the process + """ + + self._impl.detach() + + @cancellable + def resume(self) -> None: + """ + Resume session after network error + """ + + self._impl.resume() + + @cancellable + def enable_child_gating(self) -> None: + """ + Enable child gating + """ + + self._impl.enable_child_gating() + + @cancellable + def disable_child_gating(self) -> None: + """ + Disable child gating + """ + + self._impl.disable_child_gating() + + @cancellable + def create_script( + self, source: str, name: Optional[str] = None, snapshot: Optional[bytes] = None, runtime: Optional[str] = None + ) -> Script: + """ + Create a new script + """ + + kwargs = {"name": name, "snapshot": snapshot, "runtime": runtime} + _filter_missing_kwargs(kwargs) + return Script(self._impl.create_script(source, **kwargs)) + + @cancellable + def create_script_from_bytes( + self, data: bytes, name: Optional[str] = None, snapshot: Optional[bytes] = None, runtime: Optional[str] = None + ) -> Script: + """ + Create a new script from bytecode + """ + + kwargs = {"name": name, "snapshot": snapshot, "runtime": runtime} + _filter_missing_kwargs(kwargs) + return Script(self._impl.create_script_from_bytes(data, **kwargs)) + + @cancellable + def compile_script(self, source: str, name: Optional[str] = None, runtime: Optional[str] = None) -> bytes: + """ + Compile script source code to bytecode + """ + + kwargs = {"name": name, "runtime": runtime} + _filter_missing_kwargs(kwargs) + return self._impl.compile_script(source, **kwargs) + + @cancellable + def snapshot_script(self, embed_script: str, warmup_script: Optional[str], runtime: Optional[str] = None) -> bytes: + """ + Evaluate script and snapshot the resulting VM state + """ + kwargs = {"warmup_script": warmup_script, "runtime": runtime} + _filter_missing_kwargs(kwargs) + return self._impl.snapshot_script(embed_script, **kwargs) + + @cancellable + def setup_peer_connection( + self, stun_server: Optional[str] = None, relays: Optional[Sequence[_frida.Relay]] = None + ) -> None: + """ + Set up a peer connection with the target process + """ + + kwargs = {"stun_server": stun_server, "relays": relays} + _filter_missing_kwargs(kwargs) + self._impl.setup_peer_connection(**kwargs) + + @cancellable + def join_portal( + self, + address: str, + certificate: Optional[str] = None, + token: Optional[str] = None, + acl: Union[None, List[str], Tuple[str]] = None, + ) -> PortalMembership: + """ + Join a portal + """ + + kwargs: Dict[str, Any] = {"certificate": certificate, "token": token, "acl": acl} + _filter_missing_kwargs(kwargs) + return PortalMembership(self._impl.join_portal(address, **kwargs)) + + def on(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Add a signal handler + """ + + self._impl.on(signal, callback) + + def off(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Remove a signal handler + """ + + self._impl.off(signal, callback) -class EndpointParameters(object): - def __init__(self, address=None, port=None, certificate=None, origin=None, authentication=None, asset_root=None): - kw = {} +class Bus: + def __init__(self, impl: _frida.Bus) -> None: + self._impl = impl + self._on_message_callbacks: List[Callable[..., Any]] = [] - if address is not None: - kw['address'] = address + impl.on("message", self._on_message) - if port is not None: - kw['port'] = port + @cancellable + def attach(self) -> None: + """ + Attach to the bus + """ - if certificate is not None: - kw['certificate'] = certificate + self._impl.attach() - if origin is not None: - kw['origin'] = origin + def post(self, message: Any, data: Optional[Union[str, bytes]] = None) -> None: + """ + Post a JSON-encoded message to the bus + """ + + raw_message = json.dumps(message) + kwargs = {"data": data} + _filter_missing_kwargs(kwargs) + self._impl.post(raw_message, **kwargs) + + def on(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Add a signal handler + """ + + if signal == "message": + self._on_message_callbacks.append(callback) + else: + self._impl.on(signal, callback) + + def off(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Remove a signal handler + """ + + if signal == "message": + self._on_message_callbacks.remove(callback) + else: + self._impl.off(signal, callback) + + def _on_message(self, raw_message: str, data: Any) -> None: + message = json.loads(raw_message) + + for callback in self._on_message_callbacks[:]: + try: + callback(message, data) + except: + traceback.print_exc() + + +class Device: + """ + Represents a device that Frida connects to + """ + + def __init__(self, device: _frida.Device) -> None: + self.id = device.id + self.name = device.name + self.icon = device.icon + self.type = device.type + self.bus = Bus(device.bus) + + self._impl = device + + def __repr__(self) -> str: + return repr(self._impl) + + @property + def is_lost(self) -> bool: + """ + Query whether the device has been lost + """ + + return self._impl.is_lost() + + @cancellable + def query_system_parameters(self) -> Dict[str, Any]: + """ + Returns a dictionary of information about the host system + """ + + return self._impl.query_system_parameters() + + @cancellable + def get_frontmost_application(self, scope: Optional[str] = None) -> Optional[_frida.Application]: + """ + Get details about the frontmost application + """ + + kwargs = {"scope": scope} + _filter_missing_kwargs(kwargs) + return self._impl.get_frontmost_application(**kwargs) + + @cancellable + def enumerate_applications( + self, identifiers: Optional[Sequence[str]] = None, scope: Optional[str] = None + ) -> List[_frida.Application]: + """ + Enumerate applications + """ + + kwargs = {"identifiers": identifiers, "scope": scope} + _filter_missing_kwargs(kwargs) + return self._impl.enumerate_applications(**kwargs) + + @cancellable + def enumerate_processes( + self, pids: Optional[Sequence[int]] = None, scope: Optional[str] = None + ) -> List[_frida.Process]: + """ + Enumerate processes + """ + + kwargs = {"pids": pids, "scope": scope} + _filter_missing_kwargs(kwargs) + return self._impl.enumerate_processes(**kwargs) + + @cancellable + def get_process(self, process_name: str) -> _frida.Process: + """ + Get the process with the given name + :raises ProcessNotFoundError: if the process was not found or there were more than one process with the given 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) + ] + if len(matching) == 1: + return matching[0] + elif len(matching) > 1: + matches_list = ", ".join([f"{process.name} (pid: {process.pid})" for process in matching]) + raise _frida.ProcessNotFoundError(f"ambiguous name; it matches: {matches_list}") + else: + raise _frida.ProcessNotFoundError(f"unable to find process with name '{process_name}'") + + @cancellable + def enable_spawn_gating(self) -> None: + """ + Enable spawn gating + """ + + self._impl.enable_spawn_gating() + + @cancellable + def disable_spawn_gating(self) -> None: + """ + Disable spawn gating + """ + + self._impl.disable_spawn_gating() + + @cancellable + def enumerate_pending_spawn(self) -> List[_frida.Spawn]: + """ + Enumerate pending spawn + """ + + return self._impl.enumerate_pending_spawn() + + @cancellable + def enumerate_pending_children(self) -> List[_frida.Child]: + """ + Enumerate pending children + """ + + return self._impl.enumerate_pending_children() + + @cancellable + def spawn( + self, + program: Union[str, List[Union[str, bytes]], Tuple[Union[str, bytes]]], + argv: Union[None, List[Union[str, bytes]], Tuple[Union[str, bytes]]] = None, + envp: Optional[Dict[str, str]] = None, + env: Optional[Dict[str, str]] = None, + cwd: Optional[str] = None, + stdio: Optional[str] = None, + **kwargs: Any, + ) -> int: + """ + Spawn a process into an attachable state + """ + + if not isinstance(program, str): + argv = program + if isinstance(argv[0], bytes): + program = argv[0].decode() + else: + program = argv[0] + if len(argv) == 1: + argv = None + + kwargs = {"argv": argv, "envp": envp, "env": env, "cwd": cwd, "stdio": stdio, "aux": kwargs} + _filter_missing_kwargs(kwargs) + return self._impl.spawn(program, **kwargs) + + @cancellable + def input(self, target: ProcessTarget, data: bytes) -> None: + """ + Input data on stdin of a spawned process + :param target: the PID or name of the process + """ + + self._impl.input(self._pid_of(target), data) + + @cancellable + def resume(self, target: ProcessTarget) -> None: + """ + Resume a process from the attachable state + :param target: the PID or name of the process + """ + + self._impl.resume(self._pid_of(target)) + + @cancellable + def kill(self, target: ProcessTarget) -> None: + """ + Kill a process + :param target: the PID or name of the process + """ + self._impl.kill(self._pid_of(target)) + + @cancellable + def attach( + self, + target: ProcessTarget, + realm: Optional[str] = None, + persist_timeout: Optional[int] = None, + ) -> Session: + """ + Attach to a process + :param target: the PID or name of the process + """ + + kwargs = {"realm": realm, "persist_timeout": persist_timeout} + _filter_missing_kwargs(kwargs) + return Session(self._impl.attach(self._pid_of(target), **kwargs)) + + @cancellable + def inject_library_file(self, target: ProcessTarget, path: str, entrypoint: str, data: str) -> int: + """ + Inject a library file to a process + :param target: the PID or name of the process + """ + + return self._impl.inject_library_file(self._pid_of(target), path, entrypoint, data) + + @cancellable + def inject_library_blob(self, target: ProcessTarget, blob: bytes, entrypoint: str, data: str) -> int: + """ + Inject a library blob to a process + :param target: the PID or name of the process + """ + + return self._impl.inject_library_blob(self._pid_of(target), blob, entrypoint, data) + + @cancellable + def open_channel(self, address: str) -> IOStream: + """ + Open a device-specific communication channel + """ + + return IOStream(self._impl.open_channel(address)) + + @cancellable + def get_bus(self) -> Bus: + """ + Get the message bus of the device + """ + + return self.bus + + def on(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Add a signal handler + """ + + self._impl.on(signal, callback) + + def off(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Remove a signal handler + """ + + self._impl.off(signal, callback) + + def _pid_of(self, target: ProcessTarget) -> int: + if isinstance(target, str): + return self.get_process(target).pid + else: + return target + + +class DeviceManager: + def __init__(self, impl: _frida.DeviceManager) -> None: + self._impl = impl + + def __repr__(self) -> str: + return repr(self._impl) + + def get_local_device(self) -> Device: + """ + Get the local device + """ + + return self.get_device_matching(lambda d: d.type == "local", timeout=0) + + def get_remote_device(self) -> Device: + """ + Get the first remote device in the devices list + """ + + return self.get_device_matching(lambda d: d.type == "remote", timeout=0) + + def get_usb_device(self, timeout: int = 0) -> Device: + """ + Get the first device connected over USB in the devices list + """ + + return self.get_device_matching(lambda d: d.type == "usb", timeout) + + def get_device(self, id: Optional[str], timeout: int = 0) -> Device: + """ + Get a device by its id + """ + + return self.get_device_matching(lambda d: d.id == id, timeout) + + @cancellable + def get_device_matching(self, predicate: Callable[[Device], bool], timeout: int = 0) -> Device: + """ + Get device matching predicate + :param predicate: a function to filter the devices + :param timeout: operation timeout in seconds + """ + + if timeout < 0: + raw_timeout = -1 + elif timeout == 0: + raw_timeout = 0 + else: + raw_timeout = int(timeout * 1000.0) + return Device(self._impl.get_device_matching(lambda d: predicate(Device(d)), raw_timeout)) + + @cancellable + def enumerate_devices(self) -> List[Device]: + """ + Enumerate devices + """ + + return [Device(device) for device in self._impl.enumerate_devices()] + + @cancellable + def add_remote_device( + self, + address: str, + certificate: Optional[str] = None, + origin: Optional[str] = None, + token: Optional[str] = None, + keepalive_interval: Optional[int] = None, + ) -> Device: + """ + Add a remote device + """ + + kwargs: Dict[str, Any] = { + "certificate": certificate, + "origin": origin, + "token": token, + "keepalive_interval": keepalive_interval, + } + _filter_missing_kwargs(kwargs) + return Device(self._impl.add_remote_device(address, **kwargs)) + + @cancellable + def remove_remote_device(self, address: str) -> None: + """ + Remove a remote device + """ + + self._impl.remove_remote_device(address=address) + + def on(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Add a signal handler + """ + + self._impl.on(signal, callback) + + def off(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Remove a signal handler + """ + + self._impl.off(signal, callback) + + +class RPCException(Exception): + """ + Wraps remote errors from the script RPC + """ + + def __str__(self) -> str: + return str(self.args[2]) if len(self.args) >= 3 else str(self.args[0]) + + +class EndpointParameters: + def __init__( + self, + address: Optional[str] = None, + port: Optional[int] = None, + certificate: Optional[str] = None, + origin: Optional[str] = None, + authentication: Optional[Tuple[str, Union[str, Callable[[str], Any]]]] = None, + asset_root: Optional[str] = None, + ): + kwargs: Dict[str, Any] = {"address": address, "port": port, "certificate": certificate, "origin": origin} + if asset_root is not None: + kwargs["asset_root"] = str(asset_root) + _filter_missing_kwargs(kwargs) 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": + kwargs["auth_token"] = auth_data + elif auth_scheme == "callback": + if not callable(auth_data): + raise ValueError( + "Authentication data must provide a Callable if the authentication scheme is callback" + ) + kwargs["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) - - self._impl = _frida.EndpointParameters(**kw) + self._impl = _frida.EndpointParameters(**kwargs) -class PortalService(object): - def __init__(self, cluster_params=EndpointParameters(), control_params=None): +class PortalService: + def __init__( + self, + cluster_params: EndpointParameters = EndpointParameters(), + control_params: Optional[EndpointParameters] = None, + ) -> None: args = [cluster_params._impl] if control_params is not None: args.append(control_params._impl) @@ -533,58 +965,107 @@ class PortalService(object): self.device = impl.device self._impl = impl - self._on_authenticated_callbacks = [] - self._on_message_callbacks = [] + self._on_authenticated_callbacks: List[Callable[[int, Dict[str, Any]], Any]] = [] + self._on_message_callbacks: List[Callable[[int, Dict[str, Any], Any], Any]] = [] - 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): + def start(self) -> None: + """ + Start listening for incoming connections + :raises InvalidOperationError: if the service isn't stopped + :raises AddressInUseError: if the given address is already in use + """ + self._impl.start() @cancellable - def stop(self): + def stop(self) -> None: + """ + Stop listening for incoming connections, and kick any connected clients + :raises InvalidOperationError: if the service is already stopped + """ + self._impl.stop() - def post(self, connection_id, message, **kwargs): + def post(self, connection_id: int, message: Any, data: Optional[Union[str, bytes]] = None) -> None: + """ + Post a message to a specific control channel. + """ + raw_message = json.dumps(message) + kwargs = {"data": data} + _filter_missing_kwargs(kwargs) self._impl.post(connection_id, raw_message, **kwargs) - def narrowcast(self, tag, message, **kwargs): + def narrowcast(self, tag: str, message: Any, data: Optional[Union[str, bytes]] = None) -> None: + """ + Post a message to control channels with a specific tag + """ + raw_message = json.dumps(message) + kwargs = {"data": data} + _filter_missing_kwargs(kwargs) self._impl.narrowcast(tag, raw_message, **kwargs) - def broadcast(self, message, **kwargs): + def broadcast(self, message: Any, data: Optional[Union[str, bytes]] = None) -> None: + """ + Broadcast a message to all control channels + """ + raw_message = json.dumps(message) + kwargs = {"data": data} + _filter_missing_kwargs(kwargs) self._impl.broadcast(raw_message, **kwargs) - def enumerate_tags(self, connection_id): + def enumerate_tags(self, connection_id: int) -> List[str]: + """ + Enumerate tags of a specific connection + """ + return self._impl.enumerate_tags(connection_id) - def tag(self, connection_id, tag): + def tag(self, connection_id: int, tag: str) -> None: + """ + Tag a specific control channel + """ + self._impl.tag(connection_id, tag) - def untag(self, connection_id, tag): + def untag(self, connection_id: int, tag: str) -> None: + """ + Untag a specific control channel + """ + self._impl.untag(connection_id, tag) - def on(self, signal, callback): - if signal == 'authenticated': + def on(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Add a signal handler + """ + + 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': + def off(self, signal: str, callback: Callable[..., Any]) -> None: + """ + Remove a signal handler + """ + + 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) - def _on_authenticated(self, connection_id, raw_session_info): + def _on_authenticated(self, connection_id: int, raw_session_info: str) -> None: session_info = json.loads(raw_session_info) for callback in self._on_authenticated_callbacks[:]: @@ -593,7 +1074,7 @@ class PortalService(object): except: traceback.print_exc() - def _on_message(self, connection_id, raw_message, data): + def _on_message(self, connection_id: int, raw_message: str, data: Any) -> None: message = json.loads(raw_message) for callback in self._on_message_callbacks[:]: @@ -603,134 +1084,159 @@ class PortalService(object): traceback.print_exc() -class Compiler(object): - def __init__(self): +class Compiler: + def __init__(self) -> None: self._impl = _frida.Compiler(get_device_manager()._impl) - def __repr__(self): + def __repr__(self) -> str: return repr(self._impl) @cancellable - def build(self, *args, **kwargs): - return self._impl.build(*args, **kwargs) + def build( + self, + entrypoint: str, + project_root: Optional[str] = None, + source_maps: Optional[str] = None, + compression: Optional[str] = None, + ) -> str: + kwargs = {"project_root": project_root, "source_maps": source_maps, "compression": compression} + _filter_missing_kwargs(kwargs) + return self._impl.build(entrypoint, **kwargs) @cancellable - def watch(self, *args, **kwargs): - return self._impl.watch(*args, **kwargs) + def watch( + self, + entrypoint: str, + project_root: Optional[str] = None, + source_maps: Optional[str] = None, + compression: Optional[str] = None, + ) -> None: + kwargs = {"project_root": project_root, "source_maps": source_maps, "compression": compression} + _filter_missing_kwargs(kwargs) + return self._impl.watch(entrypoint, **kwargs) - def on(self, signal, callback): + def on(self, signal: str, callback: Callable[..., Any]) -> None: self._impl.on(signal, callback) - def off(self, signal, callback): + def off(self, signal: str, callback: Callable[..., Any]) -> None: self._impl.off(signal, callback) -class IOStream(object): - def __init__(self, impl): - self._impl = impl - - def __repr__(self): - return repr(self._impl) - - @property - def is_closed(self): - return self._impl.is_closed() - - @cancellable - def close(self): - self._impl.close() - - @cancellable - def read(self, count): - return self._impl.read(count) - - @cancellable - def read_all(self, count): - return self._impl.read_all(count) - - @cancellable - def write(self, data): - return self._impl.write(data) - - @cancellable - def write_all(self, data): - self._impl.write_all(data) - - -class Cancellable(object): - def __init__(self): - self._impl = _Cancellable() - - def __repr__(self): - return repr(self._impl) - - @property - def is_cancelled(self): - return self._impl.is_cancelled() - - def raise_if_cancelled(self): - self._impl.raise_if_cancelled() - - def get_pollfd(self): - return CancellablePollFD(self._impl) - - @classmethod - def get_current(cls): - return _Cancellable.get_current() - - def __enter__(self): - self._impl.push_current() - - def __exit__(self, *args): - self._impl.pop_current() - - def connect(self, callback): - return self._impl.connect(callback) - - def disconnect(self, handler_id): - self._impl.disconnect(handler_id) - - def cancel(self): - self._impl.cancel() - - -class CancellablePollFD(object): - def __init__(self, cancellable): +class CancellablePollFD: + def __init__(self, cancellable: _Cancellable) -> None: self.handle = cancellable.get_fd() - self._cancellable = cancellable + self._cancellable: Optional[_Cancellable] = cancellable - def __del__(self): + def __del__(self) -> None: self.release() - def release(self): + def release(self) -> None: if self._cancellable is not None: if self.handle != -1: self._cancellable.release_fd() self.handle = -1 self._cancellable = None - def __repr__(self): + def __repr__(self) -> str: return repr(self.handle) - def __enter__(self): + def __enter__(self) -> int: return self.handle - def __exit__(self, *args): + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc_value: Optional[BaseException], + trace: Optional[TracebackType], + ) -> None: self.release() -def make_auth_callback(callback): - def authenticate(token): +class Cancellable: + def __init__(self) -> None: + self._impl = _Cancellable() + + def __repr__(self) -> str: + return repr(self._impl) + + @property + def is_cancelled(self) -> bool: + """ + Query whether cancellable has been cancelled + """ + + return self._impl.is_cancelled() + + def raise_if_cancelled(self) -> None: + """ + Raise an exception if cancelled + :raises OperationCancelledError: + """ + + self._impl.raise_if_cancelled() + + def get_pollfd(self) -> CancellablePollFD: + return CancellablePollFD(self._impl) + + @classmethod + def get_current(cls) -> _frida.Cancellable: + """ + Get the top cancellable from the stack + """ + + return _Cancellable.get_current() + + def __enter__(self) -> None: + self._impl.push_current() + + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc_value: Optional[BaseException], + trace: Optional[TracebackType], + ) -> None: + self._impl.pop_current() + + def connect(self, callback: Callable[..., Any]) -> int: + """ + Register notification callback + :returns: the created handler id + """ + + return self._impl.connect(callback) + + def disconnect(self, handler_id: int) -> None: + """ + Unregister notification callback. + """ + + self._impl.disconnect(handler_id) + + def cancel(self) -> None: + """ + Set cancellable to cancelled + """ + + self._impl.cancel() + + +def make_auth_callback(callback: Callable[[str], Any]) -> Callable[[Any], str]: + """ + Wraps authenticated callbacks with JSON marshaling + """ + + def authenticate(token: str) -> str: session_info = callback(token) return json.dumps(session_info) + return authenticate -def _to_camel_case(name): +def _to_camel_case(name: str) -> str: result = "" uppercase_next = False for c in name: - if c == '_': + if c == "_": uppercase_next = True elif uppercase_next: result += c.upper() @@ -738,9 +1244,3 @@ def _to_camel_case(name): else: result += c.lower() return result - - -if sys.version_info[0] >= 3: - string_types = str, -else: - string_types = basestring, diff --git a/frida/py.typed b/frida/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/meson.build b/meson.build index 0c9c280..49e626a 100644 --- a/meson.build +++ b/meson.build @@ -54,7 +54,7 @@ if python_incdir == '' endif result = run_command(python, '-c', - 'import sys; sys.stdout.write("%d.%d" % (sys.version_info[0], sys.version_info[1]))', + 'import sys; sys.stdout.write(f"{sys.version_info[0]}.{sys.version_info[1]}")', check: true) python_version = result.stdout() python_name = 'python' + python_version 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 72782fa..ede543b 100755 --- a/setup.py +++ b/setup.py @@ -1,66 +1,45 @@ -# -*- coding: utf-8 -*- -from __future__ import print_function - -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 -try: - from io import BytesIO -except: - try: - from cStringIO import StringIO as BytesIO - except: - from StringIO import StringIO as BytesIO -try: - from urllib.request import urlopen, Request -except: - from urllib2 import urlopen, Request -try: - from urllib.parse import urljoin, urlparse, urlunparse -except: - from urlparse import urljoin, urlparse, urlunparse -try: - from html.parser import HTMLParser -except: - from HTMLParser import HTMLParser +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] -python_major_version = python_version[0] 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): @@ -75,33 +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" - elif python_major_version == 2: - os_version = "macosx-11.0-fat64" 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 = "{}-{}".format(os_name, arch_name) - elif system == 'FreeBSD': + os_version = f"{os_name}-{arch_name}" + elif system == "FreeBSD": os_version = "freebsd-" + platform.machine() else: raise NotImplementedError("unsupported OS") egg_path = os.path.expanduser( - "~{}frida-{}-py{}.{}-{}.egg".format(os.sep, frida_version, python_version[0], python_version[1], os_version)) + 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: @@ -119,10 +101,7 @@ class FridaPrebuiltExt(build_ext): index_url = normalize_url(index_url) frida_url = urljoin(index_url, "frida/") # slash is necessary here timeout = 20 - errmsg = ( - "unable to download it within {} seconds; " - "please download it manually to {}" - ).format("{}", egg_path) + errmsg = "unable to download it within {} seconds; " f"please download it manually to {egg_path}" print("downloading package list from", frida_url) try: @@ -133,13 +112,13 @@ class FridaPrebuiltExt(build_ext): parser = PEP503PageParser("frida", frida_version, os_version) parser.feed(links_html) - urls = [url for url in parser.urls if url.major == python_major_version] - if len(urls) == 0: - raise NotImplementedError("could not find prebuilt Frida extension; " - "prebuilds only provided for Python 2.7 and 3.4+") + if len(parser.urls) == 0: + raise NotImplementedError( + "could not find prebuilt Frida extension; " "prebuilds only provided for Python 3.4+" + ) - url = urls[0] + url = parser.urls[0] egg_url = urljoin(frida_url, url.url) try: @@ -162,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' and python_major_version >= 3: + 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{0}{1}.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) @@ -190,15 +171,16 @@ def get_index_url(): else: return index_url - print("using default index URL: {}".format(DEFAULT_INDEX_URL)) + print(f"using default index URL: {DEFAULT_INDEX_URL}") return DEFAULT_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): @@ -206,20 +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])) - if python_major_version == 2: - filename_pattern = filename_pattern.decode("utf-8") + 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): @@ -231,7 +217,7 @@ class PEP503PageParser(HTMLParser): self._path.append(Tag(tag, dict(attrs))) def handle_endtag(self, tag): - if tag == u"a": + if tag == "a": while True: if self._path.pop().tagname == tag: break @@ -240,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 == u"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): @@ -265,9 +248,7 @@ def check_pep503_hash(bytes_io, url): hashname, hashvalue = fragment.split("=") if hashname not in {"md5", "sha1", "sha224", "sha256", "sha348", "sha512"}: - raise ValueError("Unsupported hash algorithm: {}, hashvalue={}".format( - hashname, hashvalue, - )) + raise ValueError(f"Unsupported hash algorithm: {hashname}, hashvalue={hashvalue}") h = hashlib.new(hashname) for block in iter(partial(bytes_io.read, 4096), b""): # iterate until EOF @@ -278,10 +259,7 @@ def check_pep503_hash(bytes_io, url): if digest == hashvalue: return else: - raise ValueError( - "`{}` hash checking failed! Expected: {}, but got: {}".format( - hashname, hashvalue, digest) - ) + raise ValueError(f"`{hashname}` hash checking failed! Expected: {hashvalue}, but got: {digest}") if __name__ == "__main__": @@ -295,6 +273,7 @@ if __name__ == "__main__": author_email="oleavr@frida.re", url="https://frida.re", install_requires=["setuptools"], + python_requires=">=3.7", license="wxWindows Library Licence, Version 3.1", keywords="frida debugger dynamic instrumentation inject javascript windows macos linux ios iphone ipad android qnx", classifiers=[ @@ -309,23 +288,19 @@ if __name__ == "__main__": "Operating System :: MacOS :: MacOS X", "Operating System :: Microsoft :: Windows", "Operating System :: POSIX :: Linux", - "Programming Language :: Python :: 2", - "Programming Language :: Python :: 2.7", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.4", - "Programming Language :: Python :: 3.5", - "Programming Language :: Python :: 3.6", "Programming Language :: Python :: 3.7", "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", "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", "_frida"], + package_data={"frida": ["py.typed"], "_frida": ["py.typed"]}, + ext_modules=[Extension("_frida", [])], + cmdclass={"build_ext": FridaPrebuiltExt}, + zip_safe=False, ) diff --git a/src/_frida.c b/src/_frida.c index b2d8cfe..66f2f40 100644 --- a/src/_frida.c +++ b/src/_frida.c @@ -46,29 +46,18 @@ #endif #define PyUnicode_FromUTF8String(str) PyUnicode_DecodeUTF8 (str, strlen (str), "strict") -#if PY_MAJOR_VERSION >= 3 -# define MOD_INIT(name) PyMODINIT_FUNC PyInit_##name (void) -# define MOD_DEF(ob, name, doc, methods) \ +#define MOD_INIT(name) PyMODINIT_FUNC PyInit_##name (void) +#define MOD_DEF(ob, name, doc, methods) \ { \ static struct PyModuleDef moduledef = { \ PyModuleDef_HEAD_INIT, name, doc, -1, methods, }; \ ob = PyModule_Create (&moduledef); \ } -# define MOD_SUCCESS_VAL(val) val -# define MOD_ERROR_VAL NULL -# define PyRepr_FromString PyUnicode_FromString -# define PyRepr_FromFormat PyUnicode_FromFormat -# define PYFRIDA_GETARGSPEC_FUNCTION "getfullargspec" -#else -# define MOD_INIT(name) PyMODINIT_FUNC init##name (void) -# define MOD_DEF(ob, name, doc, methods) \ - ob = Py_InitModule3 (name, methods, doc); -# define MOD_SUCCESS_VAL(val) -# define MOD_ERROR_VAL -# define PyRepr_FromString PyString_FromString -# define PyRepr_FromFormat PyString_FromFormat -# define PYFRIDA_GETARGSPEC_FUNCTION "getargspec" -#endif +#define MOD_SUCCESS_VAL(val) val +#define MOD_ERROR_VAL NULL +#define PyRepr_FromString PyUnicode_FromString +#define PyRepr_FromFormat PyUnicode_FromFormat +#define PYFRIDA_GETARGSPEC_FUNCTION "getfullargspec" #if PY_VERSION_HEX >= 0x03080000 # define PYFRIDA_NO_PRINT_FUNC_OR_VECTORCALL_OFFSET 0 @@ -1938,11 +1927,7 @@ PyGObject_marshal_string (const gchar * str) static gboolean PyGObject_unmarshal_string (PyObject * value, const gchar ** str) { -#if PY_MAJOR_VERSION >= 3 *str = PyUnicode_AsUTF8 (value); -#else - *str = PyString_AsString (value); -#endif return *str != NULL; } @@ -3273,11 +3258,7 @@ PyDevice_input (PyDevice * self, PyObject * args) GBytes * data; GError * error = NULL; -#if PY_MAJOR_VERSION >= 3 if (!PyArg_ParseTuple (args, "ly#", &pid, &data_buffer, &data_size)) -#else - if (!PyArg_ParseTuple (args, "ls#", &pid, &data_buffer, &data_size)) -#endif return NULL; data = g_bytes_new (data_buffer, data_size); @@ -3429,11 +3410,7 @@ PyDevice_inject_library_blob (PyDevice * self, PyObject * args) GError * error = NULL; guint id; -#if PY_MAJOR_VERSION >= 3 if (!PyArg_ParseTuple (args, "ly#ss", &pid, &blob_buffer, &blob_size, &entrypoint, &data)) -#else - if (!PyArg_ParseTuple (args, "ls#ss", &pid, &blob_buffer, &blob_size, &entrypoint, &data)) -#endif return NULL; blob = g_bytes_new (blob_buffer, blob_size); @@ -4065,11 +4042,7 @@ PySession_create_script (PySession * self, PyObject * args, PyObject * kw) GError * error = NULL; FridaScript * handle; -#if PY_MAJOR_VERSION >= 3 if (!PyArg_ParseTupleAndKeywords (args, kw, "es|esy#z", keywords, "utf-8", &source, "utf-8", &name, &snapshot_data, &snapshot_size, &runtime_value)) -#else - if (!PyArg_ParseTupleAndKeywords (args, kw, "es|ess#z", keywords, "utf-8", &source, "utf-8", &name, &snapshot_data, &snapshot_size, &runtime_value)) -#endif return NULL; options = PySession_parse_script_options (name, snapshot_data, snapshot_size, runtime_value); @@ -4109,11 +4082,7 @@ PySession_create_script_from_bytes (PySession * self, PyObject * args, PyObject GError * error = NULL; FridaScript * handle; -#if PY_MAJOR_VERSION >= 3 if (!PyArg_ParseTupleAndKeywords (args, kw, "y#|esy#z", keywords, &data, &size, "utf-8", &name, &snapshot_data, &snapshot_size, &runtime_value)) -#else - if (!PyArg_ParseTupleAndKeywords (args, kw, "s#|ess#z", keywords, &data, &size, "utf-8", &name, &snapshot_data, &snapshot_size, &runtime_value)) -#endif return NULL; bytes = g_bytes_new (data, size); @@ -5453,7 +5422,6 @@ PyIOStream_write (PyIOStream * self, PyObject * args) GError * error = NULL; gssize bytes_written; -#if PY_MAJOR_VERSION >= 3 if (!PyArg_ParseTuple (args, "y*", &data)) return NULL; @@ -5462,24 +5430,11 @@ PyIOStream_write (PyIOStream * self, PyObject * args) PyErr_SetString (PyExc_TypeError, "expected a contiguous buffer"); return NULL; } -#else - PyObject * data_obj; - - if (!PyArg_ParseTuple (args, "O", &data_obj)) - return NULL; - - if (PyObject_GetBuffer (data_obj, &data, PyBUF_SIMPLE) != 0) - return NULL; -#endif Py_BEGIN_ALLOW_THREADS bytes_written = g_output_stream_write (self->output, data.buf, data.len, g_cancellable_get_current (), &error); Py_END_ALLOW_THREADS -#if PY_MAJOR_VERSION < 3 - PyBuffer_Release (&data); -#endif - if (error != NULL) return PyFrida_raise (error); @@ -5492,7 +5447,6 @@ PyIOStream_write_all (PyIOStream * self, PyObject * args) Py_buffer data; GError * error = NULL; -#if PY_MAJOR_VERSION >= 3 if (!PyArg_ParseTuple (args, "y*", &data)) return NULL; @@ -5501,24 +5455,11 @@ PyIOStream_write_all (PyIOStream * self, PyObject * args) PyErr_SetString (PyExc_TypeError, "expected a contiguous buffer"); return NULL; } -#else - PyObject * data_obj; - - if (!PyArg_ParseTuple (args, "O", &data_obj)) - return NULL; - - if (PyObject_GetBuffer (data_obj, &data, PyBUF_SIMPLE) != 0) - return NULL; -#endif Py_BEGIN_ALLOW_THREADS g_output_stream_write_all (self->output, data.buf, data.len, NULL, g_cancellable_get_current (), &error); Py_END_ALLOW_THREADS -#if PY_MAJOR_VERSION < 3 - PyBuffer_Release (&data); -#endif - if (error != NULL) return PyFrida_raise (error); @@ -5773,17 +5714,7 @@ PyFrida_raise (GError * error) g_string_append_unichar (message, g_unichar_tolower (g_utf8_get_char (error->message))); g_string_append (message, g_utf8_offset_to_pointer (error->message, 1)); -#if PY_MAJOR_VERSION >= 3 PyErr_SetString (exception, message->str); -#else - { - PyObject * value; - - value = PyUnicode_FromUTF8String (message->str); - PyErr_SetObject (exception, value); - Py_DECREF (value); - } -#endif g_string_free (message, TRUE); g_error_free (error); @@ -5794,11 +5725,7 @@ PyFrida_raise (GError * error) static gboolean PyFrida_is_string (PyObject * obj) { -#if PY_MAJOR_VERSION >= 3 return PyUnicode_Check (obj); -#else - return PyString_Check (obj); -#endif } static gchar * @@ -5859,10 +5786,6 @@ MOD_INIT (_frida) { PyObject * inspect, * datetime, * module; -#if PY_VERSION_HEX < 0x03070000 - PyEval_InitThreads (); -#endif - inspect = PyImport_ImportModule ("inspect"); inspect_getargspec = PyObject_GetAttrString (inspect, PYFRIDA_GETARGSPEC_FUNCTION); inspect_ismethod = PyObject_GetAttrString (inspect, "ismethod"); diff --git a/tests/__init__.py b/tests/__init__.py index d900320..3a85b97 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,4 +1,2 @@ -# -*- coding: utf-8 -*- - from .test_core import TestCore from .test_rpc import TestRpc diff --git a/tests/data/__init__.py b/tests/data/__init__.py index 35808e2..894fa90 100644 --- a/tests/data/__init__.py +++ b/tests/data/__init__.py @@ -1,20 +1,17 @@ -# -*- coding: utf-8 -*- - 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 b622f1a..3072713 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -1,12 +1,6 @@ -# -*- coding: utf-8 -*- - -import sys import threading import time -try: - import unittest2 as unittest -except: - import unittest +import unittest import frida @@ -17,39 +11,39 @@ 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') - self.assertRaisesMatching(frida.InvalidArgumentError, "device not found", get_nonexistent) + frida.get_device_manager().get_device_matching(lambda device: device.type == "lol") + + self.assertRaisesRegex(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) - self.assertRaisesMatching(frida.InvalidArgumentError, "device not found", wait_for_nonexistent) + frida.get_device_manager().get_device_matching(lambda device: device.type == "lol", timeout=0.1) + + self.assertRaisesRegex(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) cancellable.cancel() threading.Thread(target=cancel_after_100ms).start() - self.assertRaisesMatching(frida.OperationCancelledError, "operation was cancelled", wait_for_nonexistent) - - def assertRaisesMatching(self, exception, regex, operation): - m = self.assertRaisesRegex if sys.version_info[0] >= 3 else self.assertRaisesRegexp - m(exception, regex, operation) + self.assertRaisesRegex(frida.OperationCancelledError, "operation was cancelled", wait_for_nonexistent) -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 b19b65a..c85d692 100644 --- a/tests/test_pep503_page_parser.py +++ b/tests/test_pep503_page_parser.py @@ -1,54 +1,44 @@ -# coding: utf-8 - import sys -try: - import unittest2 as unittest -except: - import unittest +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
' @@ -85,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 628b97d..5370e36 100644 --- a/tests/test_rpc.py +++ b/tests/test_rpc.py @@ -1,14 +1,8 @@ -# -*- coding: utf-8 -*- - import platform import subprocess -import sys import threading import time -try: - import unittest2 as unittest -except: - import unittest +import unittest import frida @@ -16,6 +10,9 @@ from .data import target_program class TestRpc(unittest.TestCase): + target: subprocess.Popen + session: frida.core.Session + @classmethod def setUp(cls): system = platform.system() @@ -32,7 +29,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; @@ -48,21 +47,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 iter(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 @@ -71,13 +73,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 @@ -90,13 +95,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 @@ -109,13 +117,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 @@ -131,28 +142,18 @@ 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) self.assertEqual(script._pending, {}) def assertRaisesScriptDestroyed(self, operation): - self.assertRaisesMatching(frida.InvalidOperationError, "script has been destroyed", operation) + self.assertRaisesRegex(frida.InvalidOperationError, "script has been destroyed", operation) def assertRaisesOperationCancelled(self, operation): - self.assertRaisesMatching(frida.OperationCancelledError, "operation was cancelled", operation) - - def assertRaisesMatching(self, exception, regex, operation): - m = self.assertRaisesRegex if sys.version_info[0] >= 3 else self.assertRaisesRegexp - m(exception, regex, operation) + self.assertRaisesRegex(frida.OperationCancelledError, "operation was cancelled", operation) -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()