From 057d901c06acc74b3b2ebaabd35e989a7946bfd3 Mon Sep 17 00:00:00 2001 From: Yotam Nachum Date: Sat, 29 Jan 2022 12:38:29 +0200 Subject: [PATCH] Add type hinting Lots of functions were using *args and **kwargs that were kind of useless with type hinting, so I explicitly named every possible argument and its type. It breaks backward compatibility (sort of) because you can pass nonsense arguments anymore. --- frida/__init__.py | 66 ++-- frida/core.py | 962 ++++++++++++++++++++++++++-------------------- frida/py.typed | 0 setup.py | 1 + 4 files changed, 586 insertions(+), 443 deletions(-) create mode 100644 frida/py.typed diff --git a/frida/__init__.py b/frida/__init__.py index d1cc980..3b7c428 100644 --- a/frida/__init__.py +++ b/frida/__init__.py @@ -1,3 +1,5 @@ +from typing import Any, Callable, Dict, List, Optional, Tuple, Union + try: import _frida except Exception as ex: @@ -14,7 +16,7 @@ except Exception as ex: raise ex from . import core -__version__ = _frida.__version__ +__version__: str = _frida.__version__ get_device_manager = core.get_device_manager Relay = _frida.Relay @@ -40,58 +42,66 @@ 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]: + 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: + 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: + get_local_device().resume(target) -def kill(target, **kwargs): - get_local_device().kill(target, **kwargs) +def kill(target: core.ProcessTarget) -> None: + get_local_device().kill(target) -def attach(target, *args, **kwargs): - return get_local_device().attach(target, *args, **kwargs) +def attach(target: Union[int, str], realm: Optional[str] = None, persist_timeout: Optional[int] = None) -> core.Session: + 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: Union[int, str], path: str, entrypoint: str, data: str) -> int: + 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: Union[int, str], blob: bytes, entrypoint: str, data: str) -> int: + 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: + return get_device_matching(lambda d: d.type == "local", timeout=0) -def get_remote_device(**kwargs): - return get_device_matching(lambda d: d.type == "remote", timeout=0, **kwargs) +def get_remote_device() -> core.Device: + return get_device_matching(lambda d: d.type == "remote", timeout=0) -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: + return get_device_matching(lambda d: d.type == "usb", 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: + 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: + 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]: + return get_device_manager().enumerate_devices() @core.cancellable -def shutdown(): +def shutdown() -> None: get_device_manager()._impl.close() diff --git a/frida/core.py b/frida/core.py index 91f74b5..50ba990 100644 --- a/frida/core.py +++ b/frida/core.py @@ -1,10 +1,11 @@ import fnmatch 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 @@ -12,17 +13,28 @@ _device_manager = None _Cancellable = _frida.Cancellable +ProcessTarget = Union[int, str] -def get_device_manager(): + +def get_device_manager() -> "DeviceManager": global _device_manager if _device_manager is None: _device_manager = DeviceManager(_frida.DeviceManager()) return _device_manager -def cancellable(f): +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, **kwargs): + def wrapper(*args: Any, **kwargs: Any) -> R: cancellable = kwargs.pop("cancellable", None) if cancellable is not None: with cancellable: @@ -33,357 +45,160 @@ def cancellable(f): return wrapper -class DeviceManager: - def __init__(self, impl): +class IOStream: + def __init__(self, impl) -> 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: - 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: + return self._impl.is_closed() @cancellable - def query_system_parameters(self): - return self._impl.query_system_parameters() + def close(self) -> None: + 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: + 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: + 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: + 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: - 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): - 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, str): - 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 + def write_all(self, data: bytes) -> None: + self._impl.write_all(data) -class Bus: - 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: - def __init__(self, impl): +class PortalMembership: + def __init__(self, impl) -> 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: + self._impl.terminate() - @cancellable - def resume(self): - self._impl.resume() - @cancellable - def enable_child_gating(self): - self._impl.enable_child_gating() +class ScriptExports: + def __init__(self, script: "Script") -> None: + self._script = script - @cancellable - def disable_child_gating(self): - self._impl.disable_child_gating() + def __getattr__(self, name: str) -> Any: + script = self._script + js_name = _to_camel_case(name) - @cancellable - def create_script(self, *args, **kwargs): - return Script(self._impl.create_script(*args, **kwargs)) + def method(*args: Any, **kwargs: Any) -> Any: + return script._rpc_request("call", js_name, args, **kwargs) - @cancellable - def create_script_from_bytes(self, *args, **kwargs): - return Script(self._impl.create_script_from_bytes(*args, **kwargs)) + return method - @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) + def __dir__(self) -> List[str]: + return self._script.list_exports() class Script: - def __init__(self, impl): + 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) - def __repr__(self): + def __repr__(self) -> str: return repr(self._impl) @property - def is_destroyed(self): + def is_destroyed(self) -> bool: return self._impl.is_destroyed() @cancellable - def load(self): + def load(self) -> None: self._impl.load() @cancellable - def unload(self): + def unload(self) -> None: self._impl.unload() @cancellable - def eternalize(self): + def eternalize(self) -> None: self._impl.eternalize() - def post(self, message, **kwargs): + def post(self, message: Any, data: Optional[str] = None) -> None: 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: + kwargs = {"port": port} + _filter_missing_kwargs(kwargs) + self._impl.enable_debugger(**kwargs) @cancellable - def disable_debugger(self): + def disable_debugger(self) -> None: self._impl.disable_debugger() - def on(self, signal, callback): + def on(self, signal: str, callback: Callable[..., Any]) -> None: if signal == "message": self._on_message_callbacks.append(callback) else: self._impl.on(signal, callback) - def off(self, signal, callback): + def off(self, signal: str, callback: Callable[..., Any]) -> None: 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]: return self._log_handler - def set_log_handler(self, handler): + def set_log_handler(self, handler: Callable[[str, str], None]) -> None: self._log_handler = handler - def default_log_handler(self, level, text): + def default_log_handler(self, level: str, text: str) -> None: 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) @@ -415,7 +230,7 @@ class Script: return result[1] - def _on_rpc_message(self, request_id, operation, params, data): + 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: @@ -430,7 +245,7 @@ class Script: callback(value, error) - def _on_destroyed(self): + def _on_destroyed(self) -> None: while True: next_pending = None @@ -444,7 +259,7 @@ class Script: 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"] @@ -466,70 +281,383 @@ class Script: traceback.print_exc() -class RPCException(Exception): - def __str__(self): - return self.args[2] if len(self.args) >= 3 else self.args[0] - - -class ScriptExports: - 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: - def __init__(self, impl): +class Session: + def __init__(self, impl) -> None: self._impl = impl + def __repr__(self) -> str: + return repr(self._impl) + + @property + def is_detached(self) -> bool: + return self._impl.is_detached() + @cancellable - def terminate(self): - self._impl.terminate() + def detach(self) -> None: + self._impl.detach() + + @cancellable + def resume(self) -> None: + self._impl.resume() + + @cancellable + def enable_child_gating(self) -> None: + self._impl.enable_child_gating() + + @cancellable + def disable_child_gating(self) -> None: + 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: + 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: + 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: + 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: + 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: + 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: + 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: + self._impl.on(signal, callback) + + def off(self, signal: str, callback: Callable[..., Any]) -> None: + self._impl.off(signal, callback) + + +class Bus: + def __init__(self, impl) -> None: + self._impl = impl + self._on_message_callbacks: List[Callable[..., Any]] = [] + + impl.on("message", self._on_message) + + @cancellable + def attach(self) -> None: + self._impl.attach() + + def post(self, message: Any, data: Optional[Union[str, bytes]] = None) -> None: + 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: + if signal == "message": + self._on_message_callbacks.append(callback) + else: + self._impl.on(signal, callback) + + def off(self, signal: str, callback: Callable[..., Any]) -> None: + 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: + + def __init__(self, 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: + return self._impl.is_lost() + + @cancellable + def query_system_parameters(self) -> Dict[str, Any]: + return self._impl.query_system_parameters() + + @cancellable + def get_frontmost_application(self, scope: Optional[str] = None) -> Optional[_frida.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]: + 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]: + 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: + 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: + self._impl.enable_spawn_gating() + + @cancellable + def disable_spawn_gating(self) -> None: + self._impl.disable_spawn_gating() + + @cancellable + def enumerate_pending_spawn(self) -> List[_frida.Spawn]: + return self._impl.enumerate_pending_spawn() + + @cancellable + def enumerate_pending_children(self) -> List[_frida.Child]: + 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: + 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: + self._impl.input(self._pid_of(target), data) + + @cancellable + def resume(self, target: ProcessTarget) -> None: + self._impl.resume(self._pid_of(target)) + + @cancellable + def kill(self, target: ProcessTarget) -> None: + self._impl.kill(self._pid_of(target)) + + @cancellable + def attach( + self, + target: ProcessTarget, + realm: Optional[str] = None, + persist_timeout: Optional[int] = None, + ) -> Session: + 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: + 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: + return self._impl.inject_library_blob(self._pid_of(target), blob, entrypoint, data) + + @cancellable + def open_channel(self, address: str) -> IOStream: + return IOStream(self._impl.open_channel(address)) + + @cancellable + def get_bus(self) -> Bus: + return Bus(self._impl.get_bus()) + + def on(self, signal: str, callback: Callable[..., Any]) -> None: + self._impl.on(signal, callback) + + def off(self, signal: str, callback: Callable[..., Any]) -> None: + 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) -> None: + self._impl = impl + + def __repr__(self) -> str: + return repr(self._impl) + + def get_local_device(self) -> Device: + return self.get_device_matching(lambda d: d.type == "local", timeout=0) + + def get_remote_device(self) -> Device: + return self.get_device_matching(lambda d: d.type == "remote", timeout=0) + + def get_usb_device(self, timeout: int = 0) -> Device: + return self.get_device_matching(lambda d: d.type == "usb", timeout) + + def get_device(self, id: Optional[str], timeout: int = 0) -> Device: + 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: + 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]: + 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: + 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: + self._impl.remove_remote_device(address=address) + + def on(self, signal: str, callback: Callable[..., Any]) -> None: + self._impl.on(signal, callback) + + def off(self, signal: str, callback: Callable[..., Any]) -> None: + self._impl.off(signal, callback) + + +class RPCException(Exception): + 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=None, port=None, certificate=None, origin=None, authentication=None, asset_root=None): - kw = {} - - if address is not None: - kw["address"] = address - - if port is not None: - kw["port"] = port - - if certificate is not None: - kw["certificate"] = certificate - - if origin is not None: - kw["origin"] = origin + 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 + kwargs["auth_token"] = auth_data elif auth_scheme == "callback": - kw["auth_callback"] = make_auth_callback(auth_data) + 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: - def __init__(self, cluster_params=EndpointParameters(), control_params=None): + 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) @@ -537,42 +665,48 @@ class PortalService: 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) @cancellable - def start(self): + def start(self) -> None: self._impl.start() @cancellable - def stop(self): + def stop(self) -> None: 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: 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: 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: 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]: return self._impl.enumerate_tags(connection_id) - def tag(self, connection_id, tag): + def tag(self, connection_id: int, tag: str) -> None: self._impl.tag(connection_id, tag) - def untag(self, connection_id, tag): + def untag(self, connection_id: int, tag: str) -> None: self._impl.untag(connection_id, tag) - def on(self, signal, callback): + def on(self, signal: str, callback: Callable[..., Any]) -> None: if signal == "authenticated": self._on_authenticated_callbacks.append(callback) elif signal == "message": @@ -580,7 +714,7 @@ class PortalService: else: self._impl.on(signal, callback) - def off(self, signal, callback): + def off(self, signal: str, callback: Callable[..., Any]) -> None: if signal == "authenticated": self._on_authenticated_callbacks.remove(callback) elif signal == "message": @@ -588,7 +722,7 @@ class PortalService: 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[:]: @@ -597,7 +731,7 @@ class PortalService: 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[:]: @@ -608,130 +742,128 @@ class PortalService: class Compiler: - def __init__(self): + 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: - 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: - 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: - def __init__(self, cancellable): + 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: + return self._impl.is_cancelled() + + def raise_if_cancelled(self) -> None: + 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: + return self._impl.connect(callback) + + def disconnect(self, handler_id: int) -> None: + self._impl.disconnect(handler_id) + + def cancel(self) -> None: + self._impl.cancel() + + +def make_auth_callback(callback: Callable[[str], Any]) -> Callable[[Any], str]: + 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: diff --git a/frida/py.typed b/frida/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/setup.py b/setup.py index e20129f..1ca05fd 100755 --- a/setup.py +++ b/setup.py @@ -299,6 +299,7 @@ if __name__ == "__main__": "Topic :: Software Development :: Libraries :: Python Modules", ], packages=["frida"], + package_data={"frida": ["py.typed"]}, ext_modules=[Extension("_frida", [])], cmdclass={"build_ext": FridaPrebuiltExt}, zip_safe=False,