frida/frida-python
1
0 Fork 0
mirror of https://github.com/frida/frida-python.git synced 2026-11-07 09:35:57 -10:00
Code Issues Packages Projects Releases Wiki Activity

Auto-generate the bindings from Frida's .gir

Browse Source
This commit is contained in:
Ole André Vadla Ravnås authored and Ole André Vadla Ravnås committed 2026-07-16 17:07:24 +02:00
1 parent 0988dca83e
commit 2d5f9b0f7a
62 files changed
+6656 -9545
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.github/workflows/code-style.yml
+2 -3
View File
@@ -11,6 +11,5 @@ jobs:
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v4
with:
python-version: 3.8
- uses: jamescurtin/isort-action@master
- run: pip install isort
- run: isort --check-only --diff .
.github/workflows/linting.yml
+3 -2
View File
@@ -5,11 +5,12 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: lgeiger/pyflakes-action@master
- run: pip install pyflakes
- run: pyflakes $(git ls-files '*.py' | grep -v '^frida/frida_bindgen/assets/')
mypy:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: jpetrucciani/mypy-check@master
with:
mypy_flags: '--exclude examples --exclude setup'
mypy_flags: '--exclude examples --exclude setup --exclude frida/frida_bindgen/assets'
.github/workflows/tests.yml
+19
View File
@@ -0,0 +1,19 @@
name: tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
timeout-minutes: 20
steps:
- uses: actions/checkout@v7
with:
submodules: recursive
- uses: actions/setup-python@v6
with:
python-version: '3.x'
- name: Install build dependencies
run: pip install setuptools
- name: Build the extension
run: make
- name: Run the test suite
run: python3 -m unittest tests.test_bindgen -v
.gitignore
+2
View File
@@ -1,5 +1,7 @@
*.pyc
*.abi3.so
/*.egg-info/
/.cache/
/build/
/deps/
/dist/
.gitmodules
+3
View File
@@ -1,3 +1,6 @@
[submodule "releng"]
path = releng
url = https://github.com/frida/releng.git
[submodule "frida-bindgen"]
path = frida-bindgen
url = https://github.com/frida/frida-bindgen.git
MANIFEST.in
+3 -3
View File
@@ -1,8 +1,8 @@
include meson.build
include frida/meson.build
include frida/_frida/extension.c
include frida/_frida/extension.version
include frida/_frida/meson.build
include frida/extension.version
recursive-include frida/frida_bindgen *.py *.c *.h
recursive-include frida-bindgen *.py
include subprojects/frida-core.wrap
include BSDmakefile
Makefile
+1 -1
View File
@@ -8,7 +8,7 @@ all $(MAKECMDGOALS):
$(MAKECMDGOALS)
git-submodules:
@if [ ! -f releng/meson/meson.py ]; then \
@if [ ! -f releng/meson/meson.py ] || [ ! -f frida-bindgen/frida_bindgen_core/model.py ]; then \
git submodule update --init --recursive --depth 1; \
fi
-include git-submodules
frida-bindgen
Submodule
+1
Submodule frida-bindgen added at df17879d92.
frida/__init__.py
-189
View File
@@ -1,189 +0,0 @@
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
try:
from . import _frida
except Exception as ex:
print("")
print("***")
if str(ex).startswith("No module named "):
print("Frida native extension not found")
print("Please check your PYTHONPATH.")
else:
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__: str = _frida.__version__
get_device_manager = core.get_device_manager
Relay = _frida.Relay
PortalService = core.PortalService
EndpointParameters = core.EndpointParameters
Compiler = core.Compiler
PackageManager = core.PackageManager
FileMonitor = _frida.FileMonitor
Cancellable = core.Cancellable
ServerNotRunningError = _frida.ServerNotRunningError
ExecutableNotFoundError = _frida.ExecutableNotFoundError
ExecutableNotSupportedError = _frida.ExecutableNotSupportedError
ProcessNotFoundError = _frida.ProcessNotFoundError
ProcessNotRespondingError = _frida.ProcessNotRespondingError
InvalidArgumentError = _frida.InvalidArgumentError
InvalidOperationError = _frida.InvalidOperationError
PermissionDeniedError = _frida.PermissionDeniedError
AddressInUseError = _frida.AddressInUseError
TimedOutError = _frida.TimedOutError
NotSupportedError = _frida.NotSupportedError
ProtocolError = _frida.ProtocolError
TransportError = _frida.TransportError
OperationCancelledError = _frida.OperationCancelledError
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(
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: 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: core.ProcessTarget) -> None:
"""
Kill a process
:param target: the PID or name of the process
"""
get_local_device().kill(target)
def attach(
target: core.ProcessTarget,
realm: Optional[str] = None,
persist_timeout: Optional[int] = None,
exceptor: Optional[str] = None,
unwind_broker: Optional[bool] = None,
exit_monitor: Optional[bool] = None,
thread_suspend_monitor: Optional[bool] = None,
linker_notifier_offsets: Optional[Sequence[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,
exceptor=exceptor,
unwind_broker=unwind_broker,
exit_monitor=exit_monitor,
thread_suspend_monitor=thread_suspend_monitor,
linker_notifier_offsets=linker_notifier_offsets,
)
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: 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() -> core.Device:
"""
Get the local device
"""
return get_device_manager().get_local_device()
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: 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: 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: 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() -> List[core.Device]:
"""
Enumerate all the devices from the device manager
"""
return get_device_manager().enumerate_devices()
@core.cancellable
def shutdown() -> None:
"""
Shutdown the main device manager
"""
get_device_manager()._impl.close()
frida/_frida/__init__.pyi
-905
View File
@@ -1,905 +0,0 @@
from typing import Any, Callable, 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,
exceptor: Optional[str] = None,
unwind_broker: Optional[bool] = None,
exit_monitor: Optional[bool] = None,
thread_suspend_monitor: Optional[bool] = None,
linker_notifier_offsets: Optional[Sequence[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 open_service(self, address: str) -> Service:
"""
Open a device-specific service.
"""
...
def unpair(self) -> None:
"""
Unpair device.
"""
...
def override_option(self, name: str, value: Any) -> None:
"""
Override a backend-specific option.
"""
...
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 interrupt(self) -> None:
"""
Interrupt any JavaScript currently executing in 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 terminate(self) -> None:
"""
Interrupt execution and 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 Service(Object):
def activate(self) -> None:
"""
Activate the service.
"""
...
def cancel(self) -> None:
"""
Cancel the service.
"""
...
def request(self, parameters: Any) -> Any:
"""
Perform a request.
"""
...
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,
output_format: Optional[str] = None,
bundle_format: Optional[str] = None,
type_check: Optional[str] = None,
source_maps: Optional[str] = None,
compression: Optional[str] = None,
platform: Optional[str] = None,
externals: Optional[Sequence[str]] = None,
) -> str:
"""
Build an agent.
"""
...
def watch(
self,
entrypoint: str,
project_root: Optional[str] = None,
output_format: Optional[str] = None,
bundle_format: Optional[str] = None,
type_check: Optional[str] = None,
source_maps: Optional[str] = None,
compression: Optional[str] = None,
platform: Optional[str] = None,
externals: Optional[Sequence[str]] = None,
) -> None:
"""
Continuously build an agent.
"""
...
class PackageManager(Object):
@property
def registry(self) -> str:
"""
The registry being used.
"""
...
@registry.setter
def registry(self, value: str) -> None:
"""
Change the registry to use.
"""
...
def search(
self,
query: str,
offset: Optional[int] = None,
limit: Optional[int] = None,
) -> PackageSearchResult:
"""
Search for packages to install.
"""
...
def install(
self,
project_root: Optional[str] = None,
role: Optional[str] = None,
specs: Optional[Sequence[str]] = None,
omits: Optional[Sequence[str]] = None,
) -> PackageInstallResult:
"""
Install one or more packages.
"""
...
class Package(Object):
@property
def name(self) -> str:
"""
Package name.
"""
...
@property
def version(self) -> str:
"""
Package version.
"""
...
@property
def description(self) -> Optional[str]:
"""
Package description.
"""
...
@property
def url(self) -> Optional[str]:
"""
Package URL.
"""
...
class PackageSearchResult(Object):
@property
def packages(self) -> List[Package]:
"""
Batch of matching packages.
"""
...
@property
def total(self) -> int:
"""
Total matching packages.
"""
...
class PackageInstallResult(Object):
@property
def packages(self) -> List[Package]:
"""
The toplevel packages that are installed.
"""
...
__version__: str
frida/_frida/extension.c
-6323
View File
File diff suppressed because it is too large. Load diff
frida/_frida/meson.build
-21
View File
@@ -1,21 +0,0 @@
py_sources = [
'__init__.pyi',
'py.typed',
]
python.install_sources(py_sources, subdir: 'frida' / '_frida', pure: false)
extra_link_args = []
if host_os_family == 'darwin'
extra_link_args += '-Wl,-exported_symbol,_PyInit__frida'
elif host_os_family != 'windows'
extra_link_args += '-Wl,--version-script,' + meson.current_source_dir() / 'extension.version'
endif
extension = python.extension_module('_frida', 'extension.c',
limited_api: '3.7',
c_args: frida_component_cflags,
link_args: extra_link_args,
dependencies: [python_dep, frida_core_dep, os_deps],
install: true,
subdir: 'frida',
)
frida/core.py
-1861
View File
File diff suppressed because it is too large. Load diff
frida/_frida/extension.version → frida/extension.version
View File
File renamed without changes.
frida/_frida/py.typed → frida/frida_bindgen/__init__.py
View File
File renamed without changes.
frida/frida_bindgen/__main__.py
+4
View File
@@ -0,0 +1,4 @@
from .cli import main
if __name__ == "__main__":
main()
frida/frida_bindgen/assets/codegen_endpoint_parameters.c
+81
View File
@@ -0,0 +1,81 @@
static gboolean
PyGObject_unmarshal_certificate (const gchar * str,
GTlsCertificate ** certificate)
{
GError * error = NULL;
if (strchr (str, '\n') != NULL)
*certificate = g_tls_certificate_new_from_pem (str, -1, &error);
else
*certificate = g_tls_certificate_new_from_file (str, &error);
if (error != NULL)
goto propagate_error;
return TRUE;
propagate_error:
{
PyFrida_raise (g_error_new_literal (FRIDA_ERROR, FRIDA_ERROR_INVALID_ARGUMENT, error->message));
g_error_free (error);
return FALSE;
}
}
static int
PyEndpointParameters_init (PyEndpointParameters * self,
PyObject * args,
PyObject * kw)
{
int result = -1;
static char * keywords[] = { "address", "port", "certificate", "origin", "auth_service", "asset_root", NULL };
char * address = NULL;
unsigned short int port = 0;
char * certificate_value = NULL;
char * origin = NULL;
PyObject * auth_service_obj = NULL;
char * asset_root_value = NULL;
GTlsCertificate * certificate = NULL;
FridaAuthenticationService * auth_service = NULL;
GFile * asset_root = NULL;
FridaEndpointParameters * handle;
if (PyGObject_tp_init ((PyObject *) self, args, kw) < 0)
return -1;
if (!PyArg_ParseTupleAndKeywords (args, kw, "|esHesesOes", keywords,
"utf-8", &address,
&port,
"utf-8", &certificate_value,
"utf-8", &origin,
&auth_service_obj,
"utf-8", &asset_root_value))
return -1;
if (certificate_value != NULL && !PyGObject_unmarshal_certificate (certificate_value, &certificate))
goto beach;
if (auth_service_obj != NULL && auth_service_obj != Py_None)
auth_service = g_object_ref (PY_GOBJECT_HANDLE (auth_service_obj));
if (asset_root_value != NULL)
asset_root = g_file_new_for_path (asset_root_value);
handle = frida_endpoint_parameters_new (address, port, certificate, origin, auth_service, asset_root);
PyGObject_take_handle ((PyGObject *) self, handle, PYFRIDA_TYPE (EndpointParameters));
result = 0;
beach:
g_clear_object (&asset_root);
g_clear_object (&auth_service);
g_clear_object (&certificate);
PyMem_Free (asset_root_value);
PyMem_Free (origin);
PyMem_Free (certificate_value);
PyMem_Free (address);
return result;
}
frida/frida_bindgen/assets/codegen_gobject_globals.c
+5
View File
@@ -0,0 +1,5 @@
static GHashTable * pygobject_type_spec_by_type;
static GHashTable * frida_exception_by_error_code;
static PyObject * cancelled_exception;
static PyObject * inspect_getargspec;
static PyObject * inspect_ismethod;
frida/frida_bindgen/assets/codegen_gobject_methods.c
+1157
View File
File diff suppressed because it is too large. Load diff
frida/frida_bindgen/assets/codegen_gobject_prototypes.h
+45
View File
@@ -0,0 +1,45 @@
static void PyGObject_class_init (void);
static int PyGObject_init (PyGObject * self);
static void PyGObject_dealloc (PyGObject * self);
static void PyGObject_gc_dealloc (PyGObject * self);
static gpointer PyGObject_steal_handle (PyGObject * self);
static void PyGObject_register_type (GType instance_type, PyGObjectType * python_type);
static PyObject * PyGObject_marshal_object (gpointer handle, GType type);
static PyObject * PyGObject_new_take_handle (gpointer handle, const PyGObjectType * pytype);
static PyObject * PyGObject_try_get_from_handle (gpointer handle);
static PyObject * PyGObject_on (PyGObject * self, PyObject * args);
static PyObject * PyGObject_off (PyGObject * self, PyObject * args);
static gboolean PyGObject_parse_signal_method_args (PyObject * args, GType instance_type, guint * signal_id, PyObject ** callback);
static gint PyGObject_compare_signal_closure_callback (GClosure * closure, PyObject * callback);
static GClosure * PyGObject_make_closure_for_signal (PyObject * callback, guint max_arg_count);
static void PyGObjectSignalClosure_finalize (PyObject * callback);
static void PyGObjectSignalClosure_marshal (GClosure * closure, GValue * return_gvalue, guint n_param_values, const GValue * param_values, gpointer invocation_hint, gpointer marshal_data);
static PyObject * PyGObjectSignalClosure_marshal_params (const GValue * params, guint params_length);
static PyObject * PyGObject_marshal_value (const GValue * value);
static const gchar * PyGObject_class_name_from_c (const gchar * cname);
static guint PyFrida_get_max_argument_count (PyObject * callable);
static gboolean PyGObject_unmarshal_enum (const gchar * str, GType type, gpointer value);
static PyObject * PyGObject_marshal_enum (gint value, GType type);
static gboolean PyGObject_unmarshal_string (PyObject * value, gchar ** str);
static PyObject * PyGObject_marshal_string (const gchar * str);
static gboolean PyGObject_unmarshal_strv (PyObject * value, gchar *** strv, gint * length);
static PyObject * PyGObject_marshal_strv (gchar * const * strv, gint length);
static PyObject * PyGObject_marshal_bytes (GBytes * bytes);
static gboolean PyGObject_unmarshal_vardict (PyObject * value, GHashTable ** dict);
static PyObject * PyGObject_marshal_vardict (GHashTable * dict);
static gboolean PyGObject_unmarshal_variant (PyObject * value, GVariant ** variant);
static gboolean PyGObject_unmarshal_variant_from_sequence (PyObject * sequence, GVariant ** variant);
static gboolean PyGObject_unmarshal_variant_from_mapping (PyObject * mapping, GVariant ** variant);
static PyObject * PyGObject_marshal_variant (GVariant * variant);
static PyObject * PyGObject_marshal_variant_byte_array (GVariant * variant);
static PyObject * PyGObject_marshal_variant_dict (GVariant * variant);
static PyObject * PyGObject_marshal_variant_array (GVariant * variant);
static PyObject * PyGObject_marshal_variant_tuple (GVariant * variant);
static PyObject * PyFrida_raise (GError * error);
static PyObject * PyFrida_marshal_error (GError * error);
static void PyFrida_object_decref (gpointer obj);
static void PyFrida_deliver (PyObject * callback, PyObject * result, PyObject * error);
static PyObject * PyFrida_make_completion (GTask * task, PyFridaCompleteFunc complete);
static PyObject * PyFrida_complete_request (PyObject * module, PyObject * args);
static gboolean PyFrida_return_error (GTask * task, PyObject * error);
frida/frida_bindgen/assets/codegen_macros.h
+81
View File
@@ -0,0 +1,81 @@
#define PYFRIDA_TYPE(name) \
(&_PYFRIDA_TYPE_VAR (name, type))
#define PYFRIDA_TYPE_OBJECT(name) \
PYFRIDA_TYPE (name)->object
#define _PYFRIDA_TYPE_VAR(name, var) \
G_PASTE (G_PASTE (G_PASTE (Py, name), _), var)
#define PYFRIDA_DEFINE_BASETYPE(pyname, cname, destroy_func, ...) \
_PYFRIDA_DEFINE_TYPE_SLOTS (cname, __VA_ARGS__); \
_PYFRIDA_DEFINE_TYPE_SPEC (cname, pyname, Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE); \
static PyGObjectType _PYFRIDA_TYPE_VAR (cname, type) = \
{ \
.parent = NULL, \
.object = NULL, \
.destroy = destroy_func, \
}
#define PYFRIDA_DEFINE_TYPE(pyname, cname, parent_cname, destroy_func, ...) \
_PYFRIDA_DEFINE_TYPE_SLOTS (cname, __VA_ARGS__); \
_PYFRIDA_DEFINE_TYPE_SPEC (cname, pyname, Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE); \
static PyGObjectType _PYFRIDA_TYPE_VAR (cname, type) = \
{ \
.parent = PYFRIDA_TYPE (parent_cname), \
.object = NULL, \
.destroy = destroy_func, \
}
#define PYFRIDA_DEFINE_GC_TYPE(pyname, cname, parent_cname, destroy_func, ...) \
_PYFRIDA_DEFINE_TYPE_SLOTS (cname, __VA_ARGS__); \
_PYFRIDA_DEFINE_TYPE_SPEC (cname, pyname, Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE | Py_TPFLAGS_HAVE_GC); \
static PyGObjectType _PYFRIDA_TYPE_VAR (cname, type) = \
{ \
.parent = PYFRIDA_TYPE (parent_cname), \
.object = NULL, \
.destroy = destroy_func, \
}
#define PYFRIDA_REGISTER_TYPE(cname, gtype) \
G_BEGIN_DECLS \
{ \
PyGObjectType * t = PYFRIDA_TYPE (cname); \
t->object = PyType_FromSpecWithBases (&_PYFRIDA_TYPE_VAR (cname, spec), \
(t->parent != NULL) ? PyTuple_Pack (1, t->parent->object) : NULL); \
PyGObject_register_type (gtype, t); \
Py_IncRef (t->object); \
PyModule_AddObject (module, G_STRINGIFY (cname), t->object); \
} \
G_END_DECLS
#define _PYFRIDA_DEFINE_TYPE_SPEC(cname, pyname, type_flags) \
static PyType_Spec _PYFRIDA_TYPE_VAR (cname, spec) = \
{ \
.name = pyname, \
.basicsize = sizeof (G_PASTE (Py, cname)), \
.itemsize = 0, \
.flags = type_flags, \
.slots = _PYFRIDA_TYPE_VAR (cname, slots), \
}
#define _PYFRIDA_DEFINE_TYPE_SLOTS(cname, ...) \
static PyType_Slot _PYFRIDA_TYPE_VAR (cname, slots)[] = \
{ \
__VA_ARGS__ \
{ 0 }, \
}
#define PYFRIDA_DECLARE_EXCEPTION(error_code, name) \
G_STMT_START \
{ \
PyObject * exception = PyErr_NewException ("frida." name "Error", NULL, NULL); \
g_hash_table_insert (frida_exception_by_error_code, GINT_TO_POINTER (error_code), exception); \
Py_IncRef (exception); \
PyModule_AddObject (module, name "Error", exception); \
} \
G_STMT_END
#define PY_GOBJECT(o) ((PyGObject *) (o))
#define PY_GOBJECT_HANDLE(o) (PY_GOBJECT (o)->handle)
#define PY_GOBJECT_SIGNAL_CLOSURE(o) ((PyGObjectSignalClosure *) (o))
#define PyFrida_RETURN_NONE \
G_STMT_START \
{ \
Py_IncRef (Py_None); \
return Py_None; \
} \
G_STMT_END
frida/frida_bindgen/assets/codegen_structs.h
+29
View File
@@ -0,0 +1,29 @@
struct _PyGObject
{
PyObject_HEAD
gpointer handle;
const PyGObjectType * type;
GSList * signal_closures;
};
struct _PyGObjectType
{
PyGObjectType * parent;
PyObject * object;
PyGObjectInitFromHandleFunc init_from_handle;
GDestroyNotify destroy;
};
struct _PyGObjectSignalClosure
{
GClosure parent;
guint max_arg_count;
};
struct _PyFridaRequest
{
GTask * task;
PyFridaCompleteFunc complete;
};
frida/frida_bindgen/assets/codegen_typedefs.h
+6
View File
@@ -0,0 +1,6 @@
typedef struct _PyGObject PyGObject;
typedef void (* PyGObjectInitFromHandleFunc) (PyObject * self, gpointer handle);
typedef struct _PyGObjectType PyGObjectType;
typedef struct _PyGObjectSignalClosure PyGObjectSignalClosure;
typedef struct _PyFridaRequest PyFridaRequest;
typedef void (* PyFridaCompleteFunc) (GTask * task, PyObject * value, PyObject * error);
frida/frida_bindgen/assets/facade_bus.py
+23
View File
@@ -0,0 +1,23 @@
def _setup(self):
self._message_handlers = []
self._impl.on("message", self._on_message)
def on(self, signal, callback):
if signal == "message":
self._message_handlers.append(callback)
else:
self._impl.on(signal, _make_signal_handler(callback))
def off(self, signal, callback):
if signal == "message":
self._message_handlers.remove(callback)
else:
self._impl.off(signal, callback)
def _on_message(self, raw_message, data):
message = json.loads(raw_message)
for handler in list(self._message_handlers):
try:
handler(message, data)
except Exception:
traceback.print_exc()
frida/frida_bindgen/assets/facade_bus_aio.py
+27
View File
@@ -0,0 +1,27 @@
def _setup(self):
self._message_handlers = []
self._loop = asyncio.get_event_loop()
self._impl.on("message", self._on_message)
def on(self, signal, callback):
if signal == "message":
self._message_handlers.append(callback)
else:
self._impl.on(signal, _make_signal_handler(callback))
def off(self, signal, callback):
if signal == "message":
self._message_handlers.remove(callback)
else:
self._impl.off(signal, callback)
def _on_message(self, raw_message, data):
message = json.loads(raw_message)
for handler in list(self._message_handlers):
_dispatch(self._loop, self._deliver_message, handler, message, data)
def _deliver_message(self, handler, message, data):
try:
handler(message, data)
except Exception:
traceback.print_exc()
frida/frida_bindgen/assets/facade_cancellable.py
+22
View File
@@ -0,0 +1,22 @@
def __enter__(self):
stack = getattr(_current_cancellable, "stack", None)
if stack is None:
stack = []
_current_cancellable.stack = stack
stack.append(self)
return self
def __exit__(self, *exc):
_current_cancellable.stack.pop()
return False
@classmethod
def get_current(cls):
return _current_cancellable_get()
def raise_if_cancelled(self):
if self._impl.is_cancelled():
raise _frida.OperationCancelledError("operation was cancelled")
def get_pollfd(self):
return CancellablePollFD(self._impl)
frida/frida_bindgen/assets/facade_cancellable_aio.py
+18
View File
@@ -0,0 +1,18 @@
def __enter__(self):
self._token = _current_cancellable.set(self)
return self
def __exit__(self, *exc):
_current_cancellable.reset(self._token)
return False
@classmethod
def get_current(cls):
return _current_cancellable.get()
def raise_if_cancelled(self):
if self._impl.is_cancelled():
raise _frida.OperationCancelledError("operation was cancelled")
def get_pollfd(self):
return CancellablePollFD(self._impl)
frida/frida_bindgen/assets/facade_cancellable_helpers.py
+23
View File
@@ -0,0 +1,23 @@
class CancellablePollFD:
def __init__(self, cancellable):
self.handle = cancellable.get_fd()
self._cancellable = cancellable
def __del__(self):
self.release()
def release(self):
if self._cancellable is not None:
if self.handle != -1:
self._cancellable.release_fd()
self.handle = -1
self._cancellable = None
def __repr__(self):
return repr(self.handle)
def __enter__(self):
return self.handle
def __exit__(self, *exc):
self.release()
frida/frida_bindgen/assets/facade_cancellable_helpers_aio.py
+23
View File
@@ -0,0 +1,23 @@
class CancellablePollFD:
def __init__(self, cancellable):
self.handle = cancellable.get_fd()
self._cancellable = cancellable
def __del__(self):
self.release()
def release(self):
if self._cancellable is not None:
if self.handle != -1:
self._cancellable.release_fd()
self.handle = -1
self._cancellable = None
def __repr__(self):
return repr(self.handle)
def __enter__(self):
return self.handle
def __exit__(self, *exc):
self.release()
frida/frida_bindgen/assets/facade_device_manager_members.py
+9
View File
@@ -0,0 +1,9 @@
def get_device_matching(self, predicate, timeout=0):
deadline = time.monotonic() + timeout
while True:
for device in self.enumerate_devices():
if predicate(device):
return device
if time.monotonic() >= deadline:
raise _frida.InvalidArgumentError("no matching device found")
time.sleep(0.05)
frida/frida_bindgen/assets/facade_device_manager_members_aio.py
+9
View File
@@ -0,0 +1,9 @@
async def get_device_matching(self, predicate, timeout=0):
deadline = time.monotonic() + timeout
while True:
for device in await self.enumerate_devices():
if predicate(device):
return device
if time.monotonic() >= deadline:
raise _frida.InvalidArgumentError("no matching device found")
await asyncio.sleep(0.05)
frida/frida_bindgen/assets/facade_device_members.py
+25
View File
@@ -0,0 +1,25 @@
@property
def type(self):
return self.dtype
def get_bus(self):
return self.bus
def get_process(self, process_name):
process_name_lc = process_name.lower()
matching = [
process
for process in self.enumerate_processes()
if fnmatch.fnmatchcase(process.name.lower(), process_name_lc)
]
if len(matching) == 1:
return matching[0]
if len(matching) > 1:
matches = ", ".join(f"{p.name} (pid: {p.pid})" for p in matching)
raise _frida.ProcessNotFoundError(f"ambiguous name; it matches: {matches}")
raise _frida.ProcessNotFoundError(f"unable to find process with name '{process_name}'")
def _pid_of(self, target):
if isinstance(target, str):
return self.get_process(target).pid
return target
frida/frida_bindgen/assets/facade_device_members_aio.py
+25
View File
@@ -0,0 +1,25 @@
@property
def type(self):
return self.dtype
def get_bus(self):
return self.bus
async def get_process(self, process_name):
process_name_lc = process_name.lower()
matching = [
process
for process in await self.enumerate_processes()
if fnmatch.fnmatchcase(process.name.lower(), process_name_lc)
]
if len(matching) == 1:
return matching[0]
if len(matching) > 1:
matches = ", ".join(f"{p.name} (pid: {p.pid})" for p in matching)
raise _frida.ProcessNotFoundError(f"ambiguous name; it matches: {matches}")
raise _frida.ProcessNotFoundError(f"unable to find process with name '{process_name}'")
async def _pid_of(self, target):
if isinstance(target, str):
return (await self.get_process(target)).pid
return target
frida/frida_bindgen/assets/facade_interface.py
+12
View File
@@ -0,0 +1,12 @@
class _Implementation:
def _frida_dispatch(self, name, args, completion):
def run():
try:
result = _unwrap(getattr(self, name)(*[_wrap(a) for a in args]))
error = None
except Exception as e:
result = None
error = e
_frida._complete_request(completion, result, error)
threading.Thread(target=run, daemon=True).start()
frida/frida_bindgen/assets/facade_interface_aio.py
+17
View File
@@ -0,0 +1,17 @@
class _Implementation:
def _frida_dispatch(self, name, args, completion):
loop = self._loop
async def run():
try:
result = getattr(self, name)(*[_wrap(a) for a in args])
if asyncio.iscoroutine(result):
result = await result
result = _unwrap(result)
error = None
except Exception as e:
result = None
error = e
_frida._complete_request(completion, result, error)
_dispatch(loop, lambda: loop.create_task(run()))
frida/frida_bindgen/assets/facade_iostream.py
+25
View File
@@ -0,0 +1,25 @@
def close(self):
_invoke(self._impl.close_async, (0,))
def read(self, count):
return _invoke(self._impl.input_stream.read_bytes_async, (count, 0))
def read_all(self, count):
chunks = []
remaining = count
while remaining > 0:
chunk = _invoke(self._impl.input_stream.read_bytes_async, (remaining, 0))
if not chunk:
break
chunks.append(chunk)
remaining -= len(chunk)
return b"".join(chunks)
def write(self, data):
return _invoke(self._impl.output_stream.write_bytes_async, (data, 0))
def write_all(self, data):
view = memoryview(data)
while len(view) != 0:
written = _invoke(self._impl.output_stream.write_bytes_async, (bytes(view), 0))
view = view[written:]
frida/frida_bindgen/assets/facade_iostream_aio.py
+25
View File
@@ -0,0 +1,25 @@
async def close(self):
await _invoke(self._impl.close_async, (0,))
async def read(self, count):
return await _invoke(self._impl.input_stream.read_bytes_async, (count, 0))
async def read_all(self, count):
chunks = []
remaining = count
while remaining > 0:
chunk = await _invoke(self._impl.input_stream.read_bytes_async, (remaining, 0))
if not chunk:
break
chunks.append(chunk)
remaining -= len(chunk)
return b"".join(chunks)
async def write(self, data):
return await _invoke(self._impl.output_stream.write_bytes_async, (data, 0))
async def write_all(self, data):
view = memoryview(data)
while len(view) != 0:
written = await _invoke(self._impl.output_stream.write_bytes_async, (bytes(view), 0))
view = view[written:]
frida/frida_bindgen/assets/facade_portal.py
+37
View File
@@ -0,0 +1,37 @@
def _setup(self):
self._authenticated_handlers = []
self._message_handlers = []
self._impl.on("authenticated", self._on_authenticated)
self._impl.on("message", self._on_message)
def on(self, signal, callback):
if signal == "authenticated":
self._authenticated_handlers.append(callback)
elif signal == "message":
self._message_handlers.append(callback)
else:
self._impl.on(signal, _make_signal_handler(callback))
def off(self, signal, callback):
if signal == "authenticated":
self._authenticated_handlers.remove(callback)
elif signal == "message":
self._message_handlers.remove(callback)
else:
self._impl.off(signal, callback)
def _on_authenticated(self, connection_id, raw_session_info):
session_info = json.loads(raw_session_info)
for handler in list(self._authenticated_handlers):
try:
handler(connection_id, session_info)
except Exception:
traceback.print_exc()
def _on_message(self, connection_id, raw_message, data):
message = json.loads(raw_message)
for handler in list(self._message_handlers):
try:
handler(connection_id, message, data)
except Exception:
traceback.print_exc()
frida/frida_bindgen/assets/facade_portal_aio.py
+38
View File
@@ -0,0 +1,38 @@
def _setup(self):
self._authenticated_handlers = []
self._message_handlers = []
self._loop = asyncio.get_event_loop()
self._impl.on("authenticated", self._on_authenticated)
self._impl.on("message", self._on_message)
def on(self, signal, callback):
if signal == "authenticated":
self._authenticated_handlers.append(callback)
elif signal == "message":
self._message_handlers.append(callback)
else:
self._impl.on(signal, _make_signal_handler(callback))
def off(self, signal, callback):
if signal == "authenticated":
self._authenticated_handlers.remove(callback)
elif signal == "message":
self._message_handlers.remove(callback)
else:
self._impl.off(signal, callback)
def _on_authenticated(self, connection_id, raw_session_info):
session_info = json.loads(raw_session_info)
for handler in list(self._authenticated_handlers):
_dispatch(self._loop, self._deliver, handler, connection_id, session_info)
def _on_message(self, connection_id, raw_message, data):
message = json.loads(raw_message)
for handler in list(self._message_handlers):
_dispatch(self._loop, self._deliver, handler, connection_id, message, data)
def _deliver(self, handler, *args):
try:
handler(*args)
except Exception:
traceback.print_exc()
frida/frida_bindgen/assets/facade_rpc_helpers.py
+38
View File
@@ -0,0 +1,38 @@
def _to_camel_case(name):
result = ""
capitalize = False
for ch in name:
if ch == "_":
capitalize = True
elif capitalize:
result += ch.upper()
capitalize = False
else:
result += ch
return result
class RPCException(Exception):
def __str__(self):
return str(self.args[2]) if len(self.args) >= 3 else str(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):
if args and isinstance(args[-1], bytes):
params, data = list(args[:-1]), args[-1]
else:
params, data = list(args), None
return script._rpc_request(["call", js_name, params], data)
return method
def __dir__(self):
return self._script._list_exports()
frida/frida_bindgen/assets/facade_rpc_helpers_aio.py
+35
View File
@@ -0,0 +1,35 @@
def _to_camel_case(name):
result = ""
capitalize = False
for ch in name:
if ch == "_":
capitalize = True
elif capitalize:
result += ch.upper()
capitalize = False
else:
result += ch
return result
class RPCException(Exception):
def __str__(self):
return str(self.args[2]) if len(self.args) >= 3 else str(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):
if args and isinstance(args[-1], bytes):
params, data = list(args[:-1]), args[-1]
else:
params, data = list(args), None
return script._rpc_request(["call", js_name, params], data)
return method
frida/frida_bindgen/assets/facade_rpc_members.py
+111
View File
@@ -0,0 +1,111 @@
def _setup(self):
self.exports_sync = _ScriptExports(self)
self._message_handlers = []
self._log_handler = self.default_log_handler
self._pending = {}
self._next_request_id = 1
self._cond = threading.Condition()
self._impl.on("message", self._on_message)
self._impl.on("destroyed", self._on_destroyed)
@property
def exports(self):
return self.exports_sync
def on(self, signal, callback):
if signal == "message":
self._message_handlers.append(callback)
else:
self._impl.on(signal, _make_signal_handler(callback))
def off(self, signal, callback):
if signal == "message":
self._message_handlers.remove(callback)
else:
self._impl.off(signal, callback)
def get_log_handler(self):
return self._log_handler
def set_log_handler(self, handler):
self._log_handler = handler
def default_log_handler(self, level, text):
if level == "info":
print(text, file=sys.stdout)
else:
print(text, file=sys.stderr)
def list_exports_sync(self):
return self._rpc_request(["list"])
def list_exports(self):
return self.list_exports_sync()
def _list_exports(self):
return self.list_exports_sync()
def _rpc_request(self, args, data=None):
outcome = {}
cond = self._cond
def complete(value, error):
with cond:
outcome["value"] = value
outcome["error"] = error
cond.notify_all()
with cond:
request_id = self._next_request_id
self._next_request_id += 1
self._pending[request_id] = complete
if self._impl.is_destroyed():
self._on_destroyed()
else:
self.post(["frida:rpc", request_id] + list(args), data)
with cond:
while not outcome:
cond.wait()
if outcome["error"] is not None:
raise outcome["error"]
return outcome["value"]
def _on_message(self, raw_message, data):
message = json.loads(raw_message)
mtype = message["type"]
payload = message.get("payload")
if mtype == "log":
self._log_handler(message["level"], payload)
elif mtype == "send" and isinstance(payload, list) and payload[:1] == ["frida:rpc"]:
self._on_rpc_message(payload[1], payload[2], payload[3:], data)
else:
for handler in list(self._message_handlers):
try:
handler(message, data)
except Exception:
traceback.print_exc()
def _on_destroyed(self):
while True:
with self._cond:
pending_ids = list(self._pending.keys())
complete = self._pending.pop(pending_ids[0]) if pending_ids else None
if complete is None:
break
complete(None, _frida.InvalidOperationError("script has been destroyed"))
def _on_rpc_message(self, request_id, operation, params, data):
if operation not in ("ok", "error"):
return
complete = self._pending.pop(request_id, None)
if complete is None:
return
if operation == "error":
complete(None, RPCException(*params[0:3]))
elif data is not None:
complete((params[1], data) if len(params) > 1 else data, None)
else:
complete(params[0], None)
frida/frida_bindgen/assets/facade_rpc_members_aio.py
+98
View File
@@ -0,0 +1,98 @@
def _setup(self):
self.exports = _ScriptExports(self)
self._message_handlers = []
self._log_handler = self.default_log_handler
self._pending = {}
self._next_request_id = 1
self._loop = asyncio.get_event_loop()
self._impl.on("message", self._on_message)
self._impl.on("destroyed", self._on_destroyed)
def on(self, signal, callback):
if signal == "message":
self._message_handlers.append(callback)
else:
self._impl.on(signal, _make_signal_handler(callback))
def off(self, signal, callback):
if signal == "message":
self._message_handlers.remove(callback)
else:
self._impl.off(signal, callback)
def get_log_handler(self):
return self._log_handler
def set_log_handler(self, handler):
self._log_handler = handler
def default_log_handler(self, level, text):
if level == "info":
print(text, file=sys.stdout)
else:
print(text, file=sys.stderr)
async def list_exports(self):
return await self._rpc_request(["list"])
def _rpc_request(self, args, data=None):
loop = asyncio.get_running_loop()
future = loop.create_future()
def complete(value, error):
if future.done():
return
if error is not None:
future.set_exception(error)
else:
future.set_result(value)
request_id = self._next_request_id
self._next_request_id += 1
self._pending[request_id] = lambda value, error: _dispatch(loop, complete, value, error)
if self._impl.is_destroyed():
self._on_destroyed()
else:
self.post(["frida:rpc", request_id] + list(args), data)
return future
def _on_message(self, raw_message, data):
message = json.loads(raw_message)
mtype = message["type"]
payload = message.get("payload")
if mtype == "log":
_dispatch(self._loop, self._log_handler, message["level"], payload)
elif mtype == "send" and isinstance(payload, list) and payload[:1] == ["frida:rpc"]:
self._on_rpc_message(payload[1], payload[2], payload[3:], data)
else:
for handler in list(self._message_handlers):
_dispatch(self._loop, self._deliver_message, handler, message, data)
def _deliver_message(self, handler, message, data):
try:
handler(message, data)
except Exception:
traceback.print_exc()
def _on_destroyed(self):
while True:
pending_ids = list(self._pending.keys())
complete = self._pending.pop(pending_ids[0]) if pending_ids else None
if complete is None:
break
complete(None, _frida.InvalidOperationError("script has been destroyed"))
def _on_rpc_message(self, request_id, operation, params, data):
if operation not in ("ok", "error"):
return
complete = self._pending.pop(request_id, None)
if complete is None:
return
if operation == "error":
complete(None, RPCException(*params[0:3]))
elif data is not None:
complete((params[1], data) if len(params) > 1 else data, None)
else:
complete(params[0], None)
frida/frida_bindgen/assets/facade_toplevel.py
+63
View File
@@ -0,0 +1,63 @@
__version__ = _frida.__version__
def get_device_manager():
return _wrap(_frida.get_device_manager())
def get_device(id, timeout=0):
return get_device_manager().get_device_by_id(id, int(timeout * 1000))
def get_device_matching(predicate, timeout=0):
return get_device_manager().get_device_matching(predicate, timeout)
def get_local_device():
return get_device_manager().get_device_by_type("local", 0)
def get_remote_device():
return get_device_manager().get_device_by_type("remote", 0)
def get_usb_device(timeout=0):
return get_device_manager().get_device_by_type("usb", int(timeout * 1000))
def enumerate_devices():
return get_device_manager().enumerate_devices()
def query_system_parameters():
return get_local_device().query_system_parameters()
def spawn(program, argv=None, envp=None, env=None, cwd=None, stdio=None, **aux):
return get_local_device().spawn(
program, argv=argv, envp=envp, env=env, cwd=cwd, stdio=stdio, **aux
)
def resume(target):
return get_local_device().resume(target)
def kill(target):
return get_local_device().kill(target)
def attach(target, **kwargs):
return get_local_device().attach(target, **kwargs)
def inject_library_file(target, path, entrypoint, data):
return get_local_device().inject_library_file(target, path, entrypoint, data)
def inject_library_blob(target, blob, entrypoint, data):
return get_local_device().inject_library_blob(target, blob, entrypoint, data)
def shutdown():
get_device_manager().close()
frida/frida_bindgen/assets/facade_toplevel_aio.py
+64
View File
@@ -0,0 +1,64 @@
__version__ = _frida.__version__
def get_device_manager():
return _wrap(_frida.get_device_manager())
async def get_device(id, timeout=0):
return await get_device_manager().get_device_by_id(id, int(timeout * 1000))
async def get_device_matching(predicate, timeout=0):
return await get_device_manager().get_device_matching(predicate, timeout)
async def get_local_device():
return await get_device_manager().get_device_by_type("local", 0)
async def get_remote_device():
return await get_device_manager().get_device_by_type("remote", 0)
async def get_usb_device(timeout=0):
return await get_device_manager().get_device_by_type("usb", int(timeout * 1000))
async def enumerate_devices():
return await get_device_manager().enumerate_devices()
async def query_system_parameters():
return await (await get_local_device()).query_system_parameters()
async def spawn(program, argv=None, envp=None, env=None, cwd=None, stdio=None, **aux):
device = await get_local_device()
return await device.spawn(
program, argv=argv, envp=envp, env=env, cwd=cwd, stdio=stdio, **aux
)
async def resume(target):
return await (await get_local_device()).resume(target)
async def kill(target):
return await (await get_local_device()).kill(target)
async def attach(target, **kwargs):
return await (await get_local_device()).attach(target, **kwargs)
async def inject_library_file(target, path, entrypoint, data):
return await (await get_local_device()).inject_library_file(target, path, entrypoint, data)
async def inject_library_blob(target, blob, entrypoint, data):
return await (await get_local_device()).inject_library_blob(target, blob, entrypoint, data)
async def shutdown():
await get_device_manager().close()
frida/frida_bindgen/assets/module_get_device_manager.c
+13
View File
@@ -0,0 +1,13 @@
static PyObject *
PyFrida_get_device_manager (PyObject * module,
PyObject * args)
{
static PyObject * manager = NULL;
if (manager == NULL)
manager = PyObject_CallFunction ((PyObject *) PYFRIDA_TYPE_OBJECT (DeviceManager), NULL);
Py_IncRef (manager);
return manager;
}
frida/frida_bindgen/cli.py
+100
View File
@@ -0,0 +1,100 @@
from __future__ import annotations
import argparse
from io import StringIO
from pathlib import Path
from . import codegen
from .customization import load_customizations
from .loader import compute_model
def main():
run(build_arguments())
def build_arguments() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Generate the Python bindings for Frida.")
p.add_argument(
"--frida-gir",
required=True,
type=Path,
help="Path to the Frida .gir file.",
)
p.add_argument(
"--glib-gir",
required=True,
type=Path,
help="Path to the GLib .gir file.",
)
p.add_argument(
"--gobject-gir",
required=True,
type=Path,
help="Path to the GObject .gir file.",
)
p.add_argument(
"--gio-gir",
required=True,
type=Path,
help="Path to the GIO .gir file.",
)
p.add_argument(
"--output-py",
required=True,
type=Path,
help="Path to the output .py file.",
)
p.add_argument(
"--output-pyi",
required=True,
type=Path,
help="Path to the output .pyi file.",
)
p.add_argument(
"--output-aio",
required=True,
type=Path,
help="Path to the output asyncio facade .py file.",
)
p.add_argument(
"--output-c",
required=True,
type=Path,
help="Path to the output C file for the Python extension.",
)
return p.parse_args()
def run(args: argparse.Namespace) -> None:
customizations = load_customizations()
model = compute_model(args.frida_gir, args.glib_gir, args.gobject_gir, args.gio_gir, customizations)
artefacts = codegen.generate_all(model)
with OutputFile(args.output_py) as f:
f.write(artefacts["py"])
with OutputFile(args.output_aio) as f:
f.write(artefacts["aio"])
with OutputFile(args.output_pyi) as f:
f.write(artefacts["pyi"])
with OutputFile(args.output_c) as f:
f.write(artefacts["c"])
class OutputFile:
def __init__(self, output_path):
self._output_path = output_path
self._io = StringIO()
def __enter__(self):
return self._io
def __exit__(self, *exc):
result = self._io.getvalue()
if self._output_path.exists():
existing_contents = self._output_path.read_text(encoding="utf-8")
if existing_contents == result:
return False
self._output_path.write_text(result, encoding="utf-8")
return False
frida/frida_bindgen/codegen.py
+2422
View File
File diff suppressed because it is too large. Load diff
frida/frida_bindgen/customization.py
+384
View File
@@ -0,0 +1,384 @@
from __future__ import annotations
from typing import Mapping
from .model import (
ConstructorCustomizations,
CustomCode,
Customizations,
ModuleFunction,
EnumerationCustomizations,
EnumerationMemberCustomizations,
MethodCustomizations,
ObjectTypeCustomizations,
PropertyCustomizations,
SignalCustomizations,
TypeCustomizations,
)
_RESOLVE_PID = ("pid = self._pid_of(target)", "pid = await self._pid_of(target)")
def load_customizations() -> Customizations:
type_customizations: Mapping[str, TypeCustomizations] = {
"DeviceManager": ObjectTypeCustomizations(
custom_code=CustomCode(
members=(
"facade_device_manager_members.py",
"facade_device_manager_members_aio.py",
),
helpers=("facade_toplevel.py", "facade_toplevel_aio.py"),
module_functions=(
ModuleFunction(
py_name="get_device_manager",
c_symbol="PyFrida_get_device_manager",
asset="module_get_device_manager.c",
),
),
),
),
"Bus": ObjectTypeCustomizations(
provides_signals=True,
custom_code=CustomCode(
members=("facade_bus.py", "facade_bus_aio.py"),
),
methods={
"post": MethodCustomizations(
param_typings=["message", "data=None"],
custom_logic="json = _to_json(message)",
),
},
),
"PortalService": ObjectTypeCustomizations(
provides_signals=True,
custom_code=CustomCode(
members=("facade_portal.py", "facade_portal_aio.py"),
),
methods={
"post": MethodCustomizations(
param_typings=["connection_id", "message", "data=None"],
custom_logic="json = _to_json(message)",
),
"narrowcast": MethodCustomizations(
param_typings=["tag", "message", "data=None"],
custom_logic="json = _to_json(message)",
),
"broadcast": MethodCustomizations(
param_typings=["message", "data=None"],
custom_logic="json = _to_json(message)",
),
},
),
"Device": ObjectTypeCustomizations(
methods={
"spawn": MethodCustomizations(
param_typings=[
"program",
"argv=None",
"envp=None",
"env=None",
"cwd=None",
"stdio=None",
"**aux",
],
custom_logic="""\
if not isinstance(program, str):
argv = program
program = argv[0]
if len(argv) == 1:
argv = None
if isinstance(program, bytes):
program = program.decode()
options = _frida.SpawnOptions()
if argv is not None:
options.argv = [arg.decode() if isinstance(arg, bytes) else arg for arg in argv]
if envp is not None:
options.envp = _to_envp(envp)
if env is not None:
options.env = _to_envp(env)
if cwd is not None:
options.cwd = cwd
if stdio is not None:
options.stdio = stdio
if aux:
options.aux = aux
""",
),
"is_lost": MethodCustomizations(as_property=True),
"resume": MethodCustomizations(param_typings=["target"], custom_logic=_RESOLVE_PID),
"kill": MethodCustomizations(param_typings=["target"], custom_logic=_RESOLVE_PID),
"input": MethodCustomizations(param_typings=["target", "data"], custom_logic=_RESOLVE_PID),
"inject_library_file": MethodCustomizations(
param_typings=["target", "path", "entrypoint", "data"],
custom_logic=_RESOLVE_PID,
),
"inject_library_blob": MethodCustomizations(
param_typings=["target", "blob", "entrypoint", "data"],
custom_logic=_RESOLVE_PID,
),
"attach": MethodCustomizations(
param_typings=["target", "**kwargs"],
custom_logic=(
"pid = self._pid_of(target)\n" "options = _make_options(_frida.SessionOptions, kwargs)",
"pid = await self._pid_of(target)\n" "options = _make_options(_frida.SessionOptions, kwargs)",
),
),
},
custom_code=CustomCode(
members=("facade_device_members.py", "facade_device_members_aio.py"),
),
),
"Script": ObjectTypeCustomizations(
provides_signals=True,
custom_code=CustomCode(
members=("facade_rpc_members.py", "facade_rpc_members_aio.py"),
helpers=("facade_rpc_helpers.py", "facade_rpc_helpers_aio.py"),
),
methods={
"post": MethodCustomizations(
param_typings=["message", "data=None"],
custom_logic="json = _to_json(message)",
),
"is_destroyed": MethodCustomizations(as_property=True),
},
),
"Session": ObjectTypeCustomizations(
methods={
"is_detached": MethodCustomizations(as_property=True),
"create_script": MethodCustomizations(
param_typings=["source", "name=None", "snapshot=None", "runtime=None"],
custom_logic="""\
options = _frida.ScriptOptions()
if name is not None:
options.name = name
if snapshot is not None:
options.snapshot = snapshot
if runtime is not None:
options.runtime = runtime""",
),
"create_script_from_bytes": MethodCustomizations(
param_typings=["data", "name=None", "snapshot=None", "runtime=None"],
custom_logic="""\
bytes = data
options = _frida.ScriptOptions()
if name is not None:
options.name = name
if snapshot is not None:
options.snapshot = snapshot
if runtime is not None:
options.runtime = runtime""",
),
"compile_script": MethodCustomizations(
param_typings=["source", "name=None", "runtime=None"],
custom_logic="""\
options = _frida.ScriptOptions()
if name is not None:
options.name = name
if runtime is not None:
options.runtime = runtime""",
),
"snapshot_script": MethodCustomizations(
param_typings=["embed_script", "warmup_script=None", "runtime=None"],
custom_logic="""\
options = _frida.SnapshotOptions()
if warmup_script is not None:
options.warmup_script = warmup_script
if runtime is not None:
options.runtime = runtime""",
),
},
),
"Service": ObjectTypeCustomizations(
methods={
"is_closed": MethodCustomizations(as_property=True),
},
),
"EndpointParameters": ObjectTypeCustomizations(
custom_constructor="codegen_endpoint_parameters.c",
constructor=ConstructorCustomizations(
param_typings=[
"address=None",
"port=None",
"certificate=None",
"origin=None",
"authentication=None",
"asset_root=None",
"request_handler=None",
],
custom_logic="""\
auth_service = None
if authentication is not None:
if isinstance(authentication, tuple):
scheme, data = authentication
if scheme == "token":
auth_service = _frida.StaticAuthenticationService(data)
elif scheme == "callback":
if not callable(data):
raise ValueError("authentication data must be callable for the callback scheme")
service = AuthenticationService.__new__(AuthenticationService)
service.authenticate = lambda token: json.dumps(data(token))
AuthenticationService.__init__(service)
auth_service = service._impl
else:
raise ValueError("invalid authentication scheme")
else:
auth_service = _unwrap(authentication)
kwargs = {}
if address is not None:
kwargs["address"] = address
if port is not None:
kwargs["port"] = port
if certificate is not None:
kwargs["certificate"] = certificate
if origin is not None:
kwargs["origin"] = origin
if auth_service is not None:
kwargs["auth_service"] = auth_service
if asset_root is not None:
kwargs["asset_root"] = str(asset_root)
self._impl = _frida.EndpointParameters(**kwargs)
if request_handler is not None:
self._impl.request_handler = _unwrap(request_handler)
""",
),
),
"RelayKind": EnumerationCustomizations(
members={
"turn_udp": EnumerationMemberCustomizations(js_name="TurnUDP"),
"turn_tcp": EnumerationMemberCustomizations(js_name="TurnTCP"),
"turn_tls": EnumerationMemberCustomizations(js_name="TurnTLS"),
},
),
"ScriptRuntime": EnumerationCustomizations(
members={
"qjs": EnumerationMemberCustomizations(js_name="QJS"),
},
),
"ControlService": ObjectTypeCustomizations(
methods={
"get_endpoint_params": MethodCustomizations(drop=True),
},
properties={
"endpoint-params": PropertyCustomizations(drop=True),
},
),
"Injector": ObjectTypeCustomizations(drop=True),
"RpcClient": ObjectTypeCustomizations(drop=True),
"RpcPeer": ObjectTypeCustomizations(drop=True),
"Cancellable": ObjectTypeCustomizations(
custom_code=CustomCode(
members=("facade_cancellable.py", "facade_cancellable_aio.py"),
helpers=("facade_cancellable_helpers.py", "facade_cancellable_helpers_aio.py"),
),
methods={
"is_cancelled": MethodCustomizations(as_property=True),
"make_pollfd": MethodCustomizations(drop=True),
"source_new": MethodCustomizations(drop=True),
},
),
"IOStream": ObjectTypeCustomizations(
custom_code=CustomCode(
members=("facade_iostream.py", "facade_iostream_aio.py"),
),
methods={
"close": MethodCustomizations(drop=True),
"close_async": MethodCustomizations(suppress_facade=True),
"splice_async": MethodCustomizations(drop=True),
"has_pending": MethodCustomizations(drop=True),
"set_pending": MethodCustomizations(drop=True),
"clear_pending": MethodCustomizations(drop=True),
},
),
"InputStream": ObjectTypeCustomizations(
methods={
"close": MethodCustomizations(drop=True),
"read": MethodCustomizations(drop=True),
"read_async": MethodCustomizations(drop=True),
"read_all": MethodCustomizations(drop=True),
"read_all_async": MethodCustomizations(drop=True),
"read_bytes": MethodCustomizations(drop=True),
"skip": MethodCustomizations(drop=True),
"is_closed": MethodCustomizations(drop=True),
"has_pending": MethodCustomizations(drop=True),
"set_pending": MethodCustomizations(drop=True),
"clear_pending": MethodCustomizations(drop=True),
},
),
"OutputStream": ObjectTypeCustomizations(
methods={
"close": MethodCustomizations(drop=True),
"flush": MethodCustomizations(drop=True),
"write": MethodCustomizations(drop=True),
"write_async": MethodCustomizations(drop=True),
"write_all": MethodCustomizations(drop=True),
"write_all_async": MethodCustomizations(drop=True),
"write_bytes": MethodCustomizations(drop=True),
"writev": MethodCustomizations(drop=True),
"writev_async": MethodCustomizations(drop=True),
"writev_all": MethodCustomizations(drop=True),
"writev_all_async": MethodCustomizations(drop=True),
"splice": MethodCustomizations(drop=True),
"is_closing": MethodCustomizations(drop=True),
"is_closed": MethodCustomizations(drop=True),
"has_pending": MethodCustomizations(drop=True),
"set_pending": MethodCustomizations(drop=True),
"clear_pending": MethodCustomizations(drop=True),
},
),
"UnixSocketAddress": ObjectTypeCustomizations(
methods={
"get_path_len": MethodCustomizations(drop=True),
"get_is_abstract": MethodCustomizations(drop=True),
},
properties={
"abstract": PropertyCustomizations(drop=True),
"path-as-array": PropertyCustomizations(drop=True),
},
),
"SocketAddress": ObjectTypeCustomizations(
constructor=ConstructorCustomizations(drop=True),
methods={
"to_native": MethodCustomizations(drop=True),
},
),
"SocketConnectable": ObjectTypeCustomizations(drop_abstract_base=True),
"InetAddress": ObjectTypeCustomizations(
properties={
"bytes": PropertyCustomizations(drop=True),
},
),
"Object": ObjectTypeCustomizations(
methods={
"is_floating": MethodCustomizations(drop=True),
"ref": MethodCustomizations(drop=True),
"ref_sink": MethodCustomizations(drop=True),
"unref": MethodCustomizations(drop=True),
"getv": MethodCustomizations(drop=True),
"get_property": MethodCustomizations(drop=True),
"set_property": MethodCustomizations(drop=True),
"notify_by_pspec": MethodCustomizations(drop=True),
"freeze_notify": MethodCustomizations(drop=True),
"thaw_notify": MethodCustomizations(drop=True),
"bind_property": MethodCustomizations(drop=True),
"bind_property_full": MethodCustomizations(drop=True),
"bind_property_with_closures": MethodCustomizations(drop=True),
"force_floating": MethodCustomizations(drop=True),
"get_data": MethodCustomizations(drop=True),
"get_qdata": MethodCustomizations(drop=True),
"run_dispose": MethodCustomizations(drop=True),
"set_data": MethodCustomizations(drop=True),
"steal_data": MethodCustomizations(drop=True),
"steal_qdata": MethodCustomizations(drop=True),
"watch_closure": MethodCustomizations(drop=True),
},
signals={
"notify": SignalCustomizations(drop=True),
},
),
}
return Customizations(type_customizations=type_customizations)
frida/frida_bindgen/loader.py
+45
View File
@@ -0,0 +1,45 @@
from __future__ import annotations
from pathlib import Path
import frida_bindgen_core as core
from .model import FACTORY, Customizations, Model
INCLUDED_GIO_OBJECT_TYPES = [
"Cancellable",
"IOStream",
"InputStream",
"OutputStream",
"SocketAddress",
"SocketAddressEnumerator",
"SocketConnectable",
"InetSocketAddress",
"InetAddress",
"UnixSocketAddress",
]
INCLUDED_GIO_ENUMERATIONS = [
"FileMonitorEvent",
"SocketFamily",
"UnixSocketAddressType",
]
def compute_model(
frida_gir: Path,
glib_gir: Path,
gobject_gir: Path,
gio_gir: Path,
customizations: Customizations,
) -> Model:
return core.compute_model(
frida_gir,
glib_gir,
gobject_gir,
gio_gir,
customizations,
FACTORY,
INCLUDED_GIO_OBJECT_TYPES,
INCLUDED_GIO_ENUMERATIONS,
seed_object_first=True,
)
frida/frida_bindgen/model.py
+341
View File
@@ -0,0 +1,341 @@
from __future__ import annotations
from collections import OrderedDict
from dataclasses import dataclass, field
from functools import cached_property
from typing import List, Mapping, Optional, Sequence, Tuple, Union
import frida_bindgen_core as core
from frida_bindgen_core import TransferOwnership
from frida_bindgen_core.naming import to_pascal_case, to_snake_case
class Model(core.Model):
@cached_property
def regular_object_types(self) -> List[ObjectType]:
return [t for t in self.object_types.values() if not t.is_frida_list]
class ObjectType(core.ObjectType):
@cached_property
def py_name(self) -> str:
custom = self.customizations
if custom is not None and custom.py_name is not None:
return custom.py_name
return "GObject" if self.name == "Object" else self.name
@property
def custom_code(self) -> Optional[CustomCode]:
custom = self.customizations
return custom.custom_code if custom is not None else None
@property
def provides_signals(self) -> bool:
custom = self.customizations
return custom is not None and custom.provides_signals
@property
def custom_constructor(self) -> Optional[str]:
custom = self.customizations
return custom.custom_constructor if custom is not None else None
@property
def constructor_custom_params(self) -> List[str]:
custom = self.customizations
if custom is None or custom.constructor is None:
return []
return custom.constructor.param_typings or []
@property
def constructor_custom_logic(self) -> Optional[str]:
custom = self.customizations
if custom is None or custom.constructor is None:
return None
return custom.constructor.custom_logic
@cached_property
def c_symbol_prefix(self) -> str:
return f"Py{self.py_name}"
@cached_property
def parent_c_symbol_prefix(self) -> str:
parent = self.parent
return parent.c_symbol_prefix if parent is not None else "PyGObject"
class ClassObjectType(ObjectType):
pass
class InterfaceObjectType(ObjectType):
@cached_property
def has_abstract_base(self) -> bool:
custom = self.customizations
if custom is None:
return True
return not custom.drop_abstract_base
class Constructor(core.Constructor):
pass
class Method(core.Method):
@cached_property
def cself_name(self) -> str:
return to_snake_case(self.object_type.name).split("_")[-1]
@cached_property
def param_ctypings(self) -> List[str]:
result = [f"{self.object_type.c_type} * {self.cself_name}"]
result += [param.ctyping for param in self.parameters]
if self.is_async:
result += ["GAsyncReadyCallback callback", "gpointer user_data"]
return result
@cached_property
def finish_param_ctypings(self) -> List[str]:
result = [
f"{self.object_type.c_type} * {self.cself_name}",
"GAsyncResult * result",
]
if self.throws:
result.append("GError ** error")
return result
@cached_property
def operation_type_name(self) -> str:
return f"Py{self.object_type.name}{to_pascal_case(self.name)}Operation"
@cached_property
def customizations(self) -> Optional[MethodCustomizations]:
custom = self.object_type.customizations
if custom is None:
return None
return custom.methods.get(self.name)
@property
def suppress_facade(self) -> bool:
custom = self.customizations
return custom is not None and custom.suppress_facade
@property
def as_property(self) -> bool:
custom = self.customizations
return custom is not None and custom.as_property
@property
def custom_facade_params(self) -> List[str]:
custom = self.customizations
if custom is None:
return []
return custom.param_typings or []
@property
def custom_logic(self) -> Optional[Union[str, Tuple[str, str]]]:
custom = self.customizations
return custom.custom_logic if custom is not None else None
@cached_property
def facade_call_args(self) -> str:
return "".join(f"{param.name}, " for param in self.input_parameters if param.type.name != "Gio.Cancellable")
class Property(core.Property):
pass
class Signal(core.Signal):
pass
class Parameter(core.Parameter):
@cached_property
def ctyping(self) -> str:
return f"{self.type.c} {self.name}"
class ReturnValue(core.ReturnValue):
@cached_property
def destroy_func(self) -> Optional[str]:
if self.transfer_ownership == TransferOwnership.none:
return None
return self.type.destroy_func
class Enumeration(core.Enumeration):
@cached_property
def c_symbol_prefix(self) -> str:
return f"Py{self.name}"
class EnumerationMember(core.EnumerationMember):
@cached_property
def js_name(self) -> str:
custom = self.customizations
if custom is not None and custom.js_name is not None:
return custom.js_name
return to_pascal_case(self.name)
@cached_property
def nick(self) -> str:
return self.name.replace("_", "-")
@cached_property
def customizations(self) -> Optional[EnumerationMemberCustomizations]:
custom = self.enumeration.customizations
if custom is None:
return None
return custom.members.get(self.name)
@dataclass
class Customizations:
type_customizations: Mapping[str, TypeCustomizations] = field(default_factory=OrderedDict)
@dataclass
class TypeCustomizations:
pass
@dataclass
class ModuleFunction:
py_name: str
c_symbol: str
asset: str
@dataclass
class CustomCode:
members: Optional[Tuple[str, str]] = None
helpers: Optional[Tuple[str, str]] = None
module_functions: Optional[Tuple[ModuleFunction, ...]] = None
@dataclass
class ObjectTypeCustomizations(TypeCustomizations):
py_name: Optional[str] = None
drop: bool = False
drop_abstract_base: bool = False
custom_code: Optional[CustomCode] = None
provides_signals: bool = False
custom_constructor: Optional[str] = None
constructor: Optional[ConstructorCustomizations] = None
methods: Mapping[str, MethodCustomizations] = field(default_factory=dict)
properties: Mapping[str, PropertyCustomizations] = field(default_factory=dict)
signals: Mapping[str, SignalCustomizations] = field(default_factory=dict)
@dataclass
class ConstructorCustomizations:
drop: bool = False
param_typings: Optional[List[str]] = None
custom_logic: Optional[str] = None
@dataclass
class MethodCustomizations:
drop: bool = False
suppress_facade: bool = False
as_property: bool = False
param_typings: Optional[List[str]] = None
custom_logic: Optional[Union[str, Tuple[str, str]]] = None
@dataclass
class PropertyCustomizations:
drop: bool = False
@dataclass
class SignalCustomizations:
drop: bool = False
@dataclass
class EnumerationCustomizations(TypeCustomizations):
members: Mapping[str, EnumerationMemberCustomizations] = field(default_factory=dict)
@dataclass
class EnumerationMemberCustomizations:
js_name: Optional[str] = None
def _make_class(
*,
name,
c_type,
get_type,
type_struct,
parent,
constructors,
methods,
properties,
signals,
implements,
resolve_type,
model,
):
return ClassObjectType(
name,
c_type,
get_type,
type_struct,
parent,
constructors,
methods,
properties,
signals,
resolve_type,
model,
)
def _make_interface(
*,
name,
c_type,
get_type,
type_struct,
parent,
constructors,
methods,
properties,
signals,
resolve_type,
model,
):
return InterfaceObjectType(
name,
c_type,
get_type,
type_struct,
parent,
constructors,
methods,
properties,
signals,
resolve_type,
model,
)
FACTORY = core.Factory(
class_object_type=_make_class,
interface_object_type=_make_interface,
constructor=Constructor,
method=Method,
parameter=Parameter,
return_value=ReturnValue,
signal=Signal,
property_=Property,
enumeration=Enumeration,
enumeration_member=EnumerationMember,
model=Model,
)
def parse_gir(file_path: str, dependencies: Sequence[Model]) -> Model:
return core.parse_gir(file_path, dependencies, FACTORY)
frida/meson.build
+110 -3
View File
@@ -1,8 +1,115 @@
subdir('_frida')
if frida_core_dep.type_name() == 'internal'
frida_core_subprj = subproject('frida-core')
frida_gir = frida_core_subprj.get_variable('core_public_gir')
glib_gir = frida_core_subprj.get_variable('glib_gir')
gobject_gir = frida_core_subprj.get_variable('gobject_gir')
gio_gir = frida_core_subprj.get_variable('gio_gir')
else
girdir = frida_core_dep.get_variable('frida_girdir')
frida_gir = girdir / 'Frida-1.0.gir'
glib_gir = girdir / 'GLib-2.0.gir'
gobject_gir = girdir / 'GObject-2.0.gir'
gio_gir = girdir / 'Gio-2.0.gir'
endif
env = environment()
env.set('PYTHONPATH', [meson.current_source_dir(), meson.project_source_root() / 'frida-bindgen'])
code = custom_target('binding-code',
output: [
'__init__.py',
'_frida.pyi',
'extension.c',
'aio.py',
],
input: [
frida_gir,
glib_gir,
gobject_gir,
gio_gir,
files(
'frida_bindgen' / '__init__.py',
'frida_bindgen' / '__main__.py',
'frida_bindgen' / 'cli.py',
'frida_bindgen' / 'codegen.py',
'frida_bindgen' / 'customization.py',
'frida_bindgen' / 'loader.py',
'frida_bindgen' / 'model.py',
'frida_bindgen' / 'assets' / 'codegen_gobject_globals.c',
'frida_bindgen' / 'assets' / 'codegen_gobject_methods.c',
'frida_bindgen' / 'assets' / 'codegen_gobject_prototypes.h',
'frida_bindgen' / 'assets' / 'codegen_macros.h',
'frida_bindgen' / 'assets' / 'codegen_structs.h',
'frida_bindgen' / 'assets' / 'codegen_typedefs.h',
'frida_bindgen' / 'assets' / 'facade_toplevel.py',
'frida_bindgen' / 'assets' / 'facade_toplevel_aio.py',
'frida_bindgen' / 'assets' / 'facade_cancellable.py',
'frida_bindgen' / 'assets' / 'facade_cancellable_aio.py',
'frida_bindgen' / 'assets' / 'facade_cancellable_helpers.py',
'frida_bindgen' / 'assets' / 'facade_cancellable_helpers_aio.py',
'frida_bindgen' / 'assets' / 'facade_device_members.py',
'frida_bindgen' / 'assets' / 'facade_device_members_aio.py',
'frida_bindgen' / 'assets' / 'facade_device_manager_members.py',
'frida_bindgen' / 'assets' / 'facade_device_manager_members_aio.py',
'frida_bindgen' / 'assets' / 'facade_bus.py',
'frida_bindgen' / 'assets' / 'facade_bus_aio.py',
'frida_bindgen' / 'assets' / 'facade_portal.py',
'frida_bindgen' / 'assets' / 'facade_portal_aio.py',
'frida_bindgen' / 'assets' / 'facade_rpc_helpers.py',
'frida_bindgen' / 'assets' / 'facade_rpc_helpers_aio.py',
'frida_bindgen' / 'assets' / 'facade_rpc_members.py',
'frida_bindgen' / 'assets' / 'facade_rpc_members_aio.py',
'frida_bindgen' / 'assets' / 'facade_iostream.py',
'frida_bindgen' / 'assets' / 'facade_iostream_aio.py',
'frida_bindgen' / 'assets' / 'facade_interface.py',
'frida_bindgen' / 'assets' / 'facade_interface_aio.py',
'frida_bindgen' / 'assets' / 'codegen_endpoint_parameters.c',
),
files(
meson.project_source_root() / 'frida-bindgen' / 'frida_bindgen_core' / '__init__.py',
meson.project_source_root() / 'frida-bindgen' / 'frida_bindgen_core' / 'loader.py',
meson.project_source_root() / 'frida-bindgen' / 'frida_bindgen_core' / 'model.py',
meson.project_source_root() / 'frida-bindgen' / 'frida_bindgen_core' / 'naming.py',
),
],
command: [
python, '-m', 'frida_bindgen',
'--frida-gir=@INPUT0@',
'--glib-gir=@INPUT1@',
'--gobject-gir=@INPUT2@',
'--gio-gir=@INPUT3@',
'--output-py=@OUTPUT0@',
'--output-pyi=@OUTPUT1@',
'--output-c=@OUTPUT2@',
'--output-aio=@OUTPUT3@',
],
env: env,
install: true,
install_dir: [
python.get_install_dir() / 'frida',
python.get_install_dir() / 'frida',
false,
python.get_install_dir() / 'frida',
]
)
py_sources = [
'__init__.py',
'core.py',
'py.typed',
]
python.install_sources(py_sources, subdir: 'frida', pure: false)
extra_link_args = []
if host_os_family == 'darwin'
extra_link_args += '-Wl,-exported_symbol,_PyInit__frida'
elif host_os_family != 'windows'
extra_link_args += '-Wl,--version-script,' + meson.current_source_dir() / 'extension.version'
endif
extension = python.extension_module('_frida', code[2],
limited_api: '3.7',
c_args: frida_component_cflags,
link_args: extra_link_args,
dependencies: [python_dep, frida_core_dep, os_deps],
install: true,
subdir: 'frida',
)
meson.build
+6 -6
View File
@@ -32,9 +32,9 @@ endif
subdir('frida')
test('frida-python', python,
args: ['-m', 'unittest', 'discover'],
workdir: meson.current_source_dir(),
env: {'PYTHONPATH': meson.current_build_dir() / 'src'},
timeout: 30,
)
#test('frida-python', python,
# args: ['-m', 'unittest', 'discover'],
# workdir: meson.current_source_dir(),
# env: {'PYTHONPATH': meson.current_build_dir() / 'src'},
# timeout: 30,
#)
pyproject.toml
+14
View File
@@ -4,7 +4,21 @@ build-backend = "setuptools.build_meta"
[tool.black]
line-length = 120
extend-exclude = "frida/frida_bindgen/assets/"
[tool.isort]
profile = "black"
line_length = 120
extend_skip_glob = ["frida/frida_bindgen/assets/*"]
known_first_party = ["frida", "frida_bindgen_core"]
[tool.mypy]
exclude = ["frida/frida_bindgen/assets/"]
[[tool.mypy.overrides]]
module = "frida_bindgen_core.*"
ignore_missing_imports = true
[[tool.mypy.overrides]]
module = ["_frida", "frida._frida", "frida.aio", "frida.core"]
ignore_missing_imports = true
setup.py
+20 -4
View File
@@ -13,6 +13,8 @@ from setuptools.extension import Extension
SOURCE_ROOT = Path(__file__).resolve().parent
FRIDA_EXTENSION = os.environ.get("FRIDA_EXTENSION", None)
GENERATED_PACKAGE_FILES = ["__init__.py", "aio.py", "_frida.pyi"]
def main():
setup(
@@ -50,12 +52,12 @@ def main():
"Topic :: Software Development :: Debuggers",
"Topic :: Software Development :: Libraries :: Python Modules",
],
packages=["frida", "frida._frida"],
package_data={"frida": ["py.typed"], "frida._frida": ["py.typed", "__init__.pyi"]},
packages=["frida"],
package_data={"frida": ["py.typed"]},
ext_modules=[
Extension(
name="frida._frida",
sources=["frida/_frida/extension.c"],
sources=[],
py_limited_api=True,
)
],
@@ -112,6 +114,7 @@ class FridaPrebuiltExt(build_ext):
target = self.get_ext_fullpath(ext.name)
Path(target).parent.mkdir(parents=True, exist_ok=True)
shutil.copy(FRIDA_EXTENSION, target)
copy_generated_package(self.build_lib)
class FridaDemandBuiltExt(build_ext):
@@ -119,11 +122,24 @@ class FridaDemandBuiltExt(build_ext):
make = SOURCE_ROOT / "make.bat" if platform.system() == "Windows" else "make"
subprocess.run([make], check=True)
outputs = [entry for entry in (SOURCE_ROOT / "build" / "frida" / "_frida").glob("_frida.*") if entry.is_file()]
outputs = [
entry
for entry in (SOURCE_ROOT / "build" / "frida").glob("_frida.*")
if entry.is_file() and entry.suffix != ".pyi"
]
assert len(outputs) == 1
target = self.get_ext_fullpath(ext.name)
Path(target).parent.mkdir(parents=True, exist_ok=True)
shutil.copy(outputs[0], target)
copy_generated_package(self.build_lib)
def copy_generated_package(build_lib):
generated = SOURCE_ROOT / "build" / "frida"
destination = Path(build_lib) / "frida"
destination.mkdir(parents=True, exist_ok=True)
for name in GENERATED_PACKAGE_FILES:
shutil.copy(generated / name, destination / name)
if __name__ == "__main__":
tests/__init__.py
-4
View File
@@ -1,4 +0,0 @@
from .test_core import TestCore
from .test_rpc import TestRpc
__all__ = ["TestCore", "TestRpc"]
tests/compile_bindgen.sh
Executable
+38
View File
@@ -0,0 +1,38 @@
#!/bin/sh
# Quick feedback loop: regenerate bindings via run_bindgen.sh, then compile the
# generated extension.c with the exact flags ninja would use, without linking.
set -e
repo=$(cd "$(dirname "$0")/.." && pwd)
"$repo/tests/run_bindgen.sh"
python3 - "$repo" "$@" <<'EOF'
import json
import shlex
import subprocess
import sys
repo = sys.argv[1]
extra = sys.argv[2:]
cdb = json.load(open(f"{repo}/build/compile_commands.json"))
entry = next(e for e in cdb if e["file"].endswith("extension.c"))
args = shlex.split(entry["command"])
cmd = []
i = 0
while i < len(args):
a = args[i]
if a in ("-o", "-MF", "-MQ", "-c"):
i += 2
continue
if a == "-MD":
i += 1
continue
cmd.append(a)
i += 1
cmd += ["-fsyntax-only", f"{repo}/build/bindgen-out/extension.c"] + extra
sys.exit(subprocess.run(cmd, cwd=entry["directory"]).returncode)
EOF
tests/run_bindgen.sh
Executable
+20
View File
@@ -0,0 +1,20 @@
#!/bin/sh
# Quick feedback loop: run the bindgen against the .gir files from the build
# tree, without going through ninja. Outputs land in tests/bindgen-out/.
set -e
repo=$(dirname "$0")/..
girdir=$repo/build/subprojects/frida-core/src/api
outdir=$repo/build/bindgen-out
mkdir -p "$outdir"
PYTHONPATH=$repo/frida:$repo/frida-bindgen exec python3 -m frida_bindgen \
--frida-gir="$girdir/Frida-1.0.gir" \
--glib-gir="$girdir/GLib-2.0.gir" \
--gobject-gir="$girdir/GObject-2.0.gir" \
--gio-gir="$girdir/Gio-2.0.gir" \
--output-py="$outdir/__init__.py" \
--output-aio="$outdir/aio.py" \
--output-pyi="$outdir/_frida.pyi" \
--output-c="$outdir/extension.c" \
"$@"
tests/test_bindgen.py
+934
View File
@@ -0,0 +1,934 @@
import os
import socket
import subprocess
import sys
import unittest
from pathlib import Path
def free_port():
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
REPO = Path(__file__).resolve().parent.parent
BINDGEN_CORE = REPO / "frida-bindgen"
GIRDIR = REPO / "build" / "subprojects" / "frida-core" / "src" / "api"
OUTDIR = REPO / "build" / "bindgen-out"
BUILT_EXTENSION = REPO / "build" / "frida" / "_frida.abi3.so"
@unittest.skipUnless(
(GIRDIR / "Frida-1.0.gir").exists(),
"requires a configured build tree with generated .gir files",
)
class TestBindgen(unittest.TestCase):
def test_generates_all_outputs(self):
OUTDIR.mkdir(exist_ok=True)
subprocess.run(
[
sys.executable,
"-m",
"frida_bindgen",
f"--frida-gir={GIRDIR / 'Frida-1.0.gir'}",
f"--glib-gir={GIRDIR / 'GLib-2.0.gir'}",
f"--gobject-gir={GIRDIR / 'GObject-2.0.gir'}",
f"--gio-gir={GIRDIR / 'Gio-2.0.gir'}",
f"--output-py={OUTDIR / '__init__.py'}",
f"--output-aio={OUTDIR / 'aio.py'}",
f"--output-pyi={OUTDIR / '_frida.pyi'}",
f"--output-c={OUTDIR / 'extension.c'}",
],
check=True,
env={"PYTHONPATH": os.pathsep.join([str(REPO / "frida"), str(BINDGEN_CORE)])},
)
for name in ("__init__.py", "aio.py", "_frida.pyi", "extension.c"):
self.assertGreater((OUTDIR / name).stat().st_size, 0)
def test_generated_python_compiles(self):
for name in ("__init__.py", "aio.py"):
path = OUTDIR / name
if not path.exists():
self.skipTest("run test_generates_all_outputs first")
compile(path.read_text(), str(path), "exec")
@unittest.skipUnless(
BUILT_EXTENSION.exists(),
"requires a built _frida extension (run `make`)",
)
class TestExtension(unittest.TestCase):
def setUp(self):
sys.path.insert(0, str(BUILT_EXTENSION.parent))
import _frida
self._frida = _frida
def tearDown(self):
sys.path.remove(str(BUILT_EXTENSION.parent))
def test_exposes_version(self):
self.assertRegex(self._frida.__version__, r"\d+\.\d+")
def test_constructs_object_with_string_and_enum_parameters(self):
relay = self._frida.Relay("1.2.3.4:5", "user", "pass", "turn-udp")
self.assertIsInstance(relay, self._frida.Relay)
def test_constructs_object_without_parameters(self):
self.assertIsInstance(self._frida.DeviceManager(), self._frida.DeviceManager)
def test_rejects_invalid_enum_value(self):
with self.assertRaises(ValueError):
self._frida.Relay("1.2.3.4:5", "user", "pass", "bogus-kind")
def test_reads_back_string_and_enum_properties(self):
relay = self._frida.Relay("1.2.3.4:5", "bob", "secret", "turn-tcp")
self.assertEqual(relay.address, "1.2.3.4:5")
self.assertEqual(relay.username, "bob")
self.assertEqual(relay.password, "secret")
self.assertEqual(relay.kind, "turn-tcp")
def test_invokes_synchronous_methods(self):
cancellable = self._frida.Cancellable()
self.assertFalse(cancellable.is_cancelled())
cancellable.cancel()
self.assertTrue(cancellable.is_cancelled())
def test_registers_frida_exceptions(self):
for name in (
"ServerNotRunningError",
"ProcessNotFoundError",
"InvalidArgumentError",
"TransportError",
"OperationCancelledError",
):
exception = getattr(self._frida, name)
self.assertTrue(issubclass(exception, Exception))
self.assertEqual(exception.__module__, "frida")
def test_throwing_method_raises_mapped_exception(self):
cancellable = self._frida.Cancellable()
cancellable.cancel()
with self.assertRaises(self._frida.OperationCancelledError):
cancellable.set_error_if_cancelled()
def test_instantiates_handle_only_types(self):
# Types with no .gir constructor are created by the marshalling layer
# and must accept a bare, argument-free construction.
self.assertIsInstance(self._frida.Device(), self._frida.Device)
self.assertIsInstance(self._frida.Session(), self._frida.Session)
def test_exposes_object_typed_property(self):
self.assertTrue(hasattr(self._frida.IOStream, "input_stream"))
self.assertTrue(hasattr(self._frida.IOStream, "output_stream"))
def test_string_array_property_round_trips(self):
options = self._frida.SpawnOptions()
self.assertIsNone(options.argv)
options.argv = ["/bin/ls", "-la"]
self.assertEqual(options.argv, ["/bin/ls", "-la"])
options.argv = None
self.assertIsNone(options.argv)
def test_vardict_property_round_trips(self):
options = self._frida.SpawnOptions()
options.aux = {"name": "value", "count": 42, "flag": True}
self.assertEqual(options.aux, {"name": "value", "count": 42, "flag": True})
def test_variant_marshalling_types_and_casts(self):
options = self._frida.SpawnOptions()
options.aux = {
"int": 42,
"uint64": ("uint64", 0xFFFFFFFFFFFFFFFF),
"float": 3.5,
"string": "hi",
"bool": True,
"bytes": b"\x00\x01\x02",
"list": [1, 2, 3],
"nested": {"x": 1, "y": "z"},
"tuple": ("a", 1),
}
aux = options.aux
self.assertEqual(aux["int"], 42)
# the ("uint64", ...) cast preserves the value; as int64 it would be -1
self.assertEqual(aux["uint64"], 0xFFFFFFFFFFFFFFFF)
self.assertEqual(aux["float"], 3.5)
self.assertEqual(aux["string"], "hi")
self.assertIs(aux["bool"], True)
self.assertEqual(aux["bytes"], b"\x00\x01\x02")
self.assertEqual(aux["list"], [1, 2, 3])
self.assertEqual(aux["nested"], {"x": 1, "y": "z"})
self.assertEqual(aux["tuple"], ("a", 1))
def test_async_method_delivers_result_to_callback(self):
import threading
manager = self._frida.DeviceManager()
done = threading.Event()
box = {}
def on_complete(result, error):
box["result"] = result
box["error"] = error
done.set()
manager.close(on_complete)
self.assertTrue(done.wait(10), "callback was not invoked")
self.assertIsNone(box["error"])
self.assertIsNone(box["result"])
def test_raising_async_callback_does_not_corrupt_thread(self):
import contextlib
import io
import threading
raised = threading.Event()
def bad_callback(result, error):
raised.set()
raise RuntimeError("handler blew up")
with contextlib.redirect_stderr(io.StringIO()):
self._frida.DeviceManager().close(bad_callback)
self.assertTrue(raised.wait(10))
# A subsequent operation must still complete cleanly.
box = self._await_async(lambda cb: self._frida.DeviceManager().close(cb))
self.assertIsNone(box["error"])
def _await_async(self, invoke):
import threading
done = threading.Event()
box = {}
def on_complete(result, error):
box["result"] = result
box["error"] = error
done.set()
invoke(on_complete)
self.assertTrue(done.wait(10), "callback was not invoked")
return box
def test_async_method_passes_parameters_and_maps_errors(self):
manager = self._frida.DeviceManager()
box = self._await_async(lambda cb: manager.get_device_by_id("no-such-device", 1, cb))
self.assertIsNone(box["result"])
self.assertIsInstance(box["error"], self._frida.InvalidArgumentError)
def test_async_method_honors_cancellable(self):
manager = self._frida.DeviceManager()
cancellable = self._frida.Cancellable()
cancellable.cancel()
box = self._await_async(lambda cb: manager.get_device_by_id("anything", 5000, cb, cancellable))
self.assertIsInstance(box["error"], self._frida.OperationCancelledError)
@unittest.skipUnless(
(REPO / "build" / "frida" / "__init__.py").exists(),
"requires a built facade (run `make`)",
)
class TestFacade(unittest.TestCase):
def setUp(self):
sys.path.insert(0, str(REPO / "build"))
self._purge_frida()
import frida
self.frida = frida
def tearDown(self):
sys.path.remove(str(REPO / "build"))
self._purge_frida()
@staticmethod
def _purge_frida():
# The source tree's frida/ is a namespace package the stale test_core
# imports; drop it so the built facade under build/ is what loads.
for name in [n for n in sys.modules if n == "frida" or n.startswith("frida.")]:
del sys.modules[name]
def test_reexports_exceptions(self):
self.assertTrue(issubclass(self.frida.ServerNotRunningError, Exception))
def test_synchronous_method_blocks(self):
self.frida.DeviceManager().close()
def test_synchronous_method_raises_mapped_error(self):
with self.assertRaises(self.frida.InvalidArgumentError):
self.frida.DeviceManager().get_device_by_id("no-such-device", 1)
def test_aio_awaitable_completes(self):
import asyncio
import frida.aio
asyncio.run(frida.aio.DeviceManager().close())
def test_aio_awaitable_raises_mapped_error(self):
import asyncio
import frida.aio
async def scenario():
with self.assertRaises(self.frida.InvalidArgumentError):
await frida.aio.DeviceManager().get_device_by_id("no-such-device", 1)
asyncio.run(scenario())
def test_aio_cancellation_propagates(self):
import asyncio
import frida.aio
async def scenario():
manager = frida.aio.DeviceManager()
task = asyncio.ensure_future(manager.get_device_by_id("nope", 60000))
await asyncio.sleep(0.1)
task.cancel()
with self.assertRaises(asyncio.CancelledError):
await task
asyncio.run(scenario())
def test_unreferenced_interface_impl_is_collected(self):
import gc
import weakref
class MyAuth(self.frida.AuthenticationService):
def authenticate(self, token):
return "{}"
refs = [weakref.ref(MyAuth()) for _ in range(3)]
gc.collect()
self.assertTrue(all(ref() is None for ref in refs))
def test_interface_impl_survives_gc_while_referenced_by_core(self):
import gc
class MyAuth(self.frida.AuthenticationService):
def authenticate(self, token):
return "{}"
params = self.frida.EndpointParameters(address="127.0.0.1", port=0, authentication=MyAuth())
gc.collect()
gc.collect()
self.assertIsNotNone(params._impl.auth_service)
def test_interface_can_be_implemented(self):
import json
class MyAuth(self.frida.AuthenticationService):
def authenticate(self, token):
return json.dumps({"token": token})
service = MyAuth()
self.assertIsInstance(service, self.frida.AuthenticationService)
self.assertEqual(type(service._impl).__name__, "AuthenticationService")
def test_endpoint_parameters_authentication_schemes(self):
self.frida.EndpointParameters(address="::1", authentication=("token", "secret"))
self.frida.EndpointParameters(address="::1", authentication=("callback", lambda token: {"token": token}))
class MyAuth(self.frida.AuthenticationService):
def authenticate(self, token):
return "{}"
self.frida.EndpointParameters(address="::1", authentication=MyAuth())
with self.assertRaises(ValueError):
self.frida.EndpointParameters(authentication=("bogus", "x"))
def test_object_constructor_parameters(self):
cluster = self.frida.EndpointParameters(address="127.0.0.1", port=0)
service = self.frida.PortalService(cluster, None)
self.assertIsInstance(service.device, self.frida.Device)
def test_implemented_interface_receives_dispatch(self):
frida = self.frida
calls = []
def authenticate(token):
calls.append(token)
if token != "secret":
raise ValueError("wrong token")
return "{}"
control_port = free_port()
control = frida.EndpointParameters(
address="127.0.0.1", port=control_port, authentication=("callback", authenticate)
)
cluster = frida.EndpointParameters(address="127.0.0.1", port=free_port())
service = frida.PortalService(cluster, control)
service.start()
try:
with self.assertRaises(frida.InvalidArgumentError):
frida.DeviceManager().add_remote_device(
f"127.0.0.1:{control_port}", token="wrong"
).enumerate_processes()
frida.DeviceManager().add_remote_device(f"127.0.0.1:{control_port}", token="secret").enumerate_processes()
self.assertEqual(calls, ["wrong", "secret"])
finally:
service.stop()
def test_aio_implemented_interface_receives_dispatch(self):
import asyncio
import frida.aio
calls = []
async def scenario():
class MyAuth(frida.aio.AuthenticationService):
async def authenticate(self, token):
calls.append(token)
await asyncio.sleep(0)
if token != "secret":
raise ValueError("wrong token")
return "{}"
control_port = free_port()
control = frida.aio.EndpointParameters(address="127.0.0.1", port=control_port, authentication=MyAuth())
cluster = frida.aio.EndpointParameters(address="127.0.0.1", port=free_port())
service = frida.aio.PortalService(cluster, control)
await service.start()
try:
with self.assertRaises(self.frida.InvalidArgumentError):
manager = frida.aio.DeviceManager()
device = await manager.add_remote_device(f"127.0.0.1:{control_port}", token="wrong")
await device.enumerate_processes()
manager = frida.aio.DeviceManager()
device = await manager.add_remote_device(f"127.0.0.1:{control_port}", token="secret")
await device.enumerate_processes()
self.assertEqual(calls, ["wrong", "secret"])
finally:
await service.stop()
asyncio.run(scenario())
def test_web_request_handler_serves_response(self):
import urllib.request
frida = self.frida
seen = []
class Handler(frida.WebRequestHandler):
def handle_request(self, request):
seen.append(request.path)
return frida.WebResponse(200, b"pong")
control_port = free_port()
control = frida.EndpointParameters(address="127.0.0.1", port=control_port, request_handler=Handler())
cluster = frida.EndpointParameters(address="127.0.0.1", port=free_port())
service = frida.PortalService(cluster, control)
service.start()
try:
with urllib.request.urlopen(f"http://127.0.0.1:{control_port}/ping") as response:
status = response.status
body = response.read()
finally:
service.stop()
self.assertEqual(status, 200)
self.assertEqual(body, b"pong")
self.assertEqual(seen, ["/ping"])
def test_web_request_handler_returning_none_yields_not_found(self):
import urllib.error
import urllib.request
frida = self.frida
class Handler(frida.WebRequestHandler):
def handle_request(self, request):
return None
control_port = free_port()
control = frida.EndpointParameters(address="127.0.0.1", port=control_port, request_handler=Handler())
cluster = frida.EndpointParameters(address="127.0.0.1", port=free_port())
service = frida.PortalService(cluster, control)
service.start()
try:
with self.assertRaises(urllib.error.HTTPError) as raised:
urllib.request.urlopen(f"http://127.0.0.1:{control_port}/ping")
self.assertEqual(raised.exception.code, 404)
finally:
service.stop()
def test_web_request_handler_error_propagates(self):
import urllib.error
import urllib.request
frida = self.frida
class Handler(frida.WebRequestHandler):
def handle_request(self, request):
raise ValueError("boom")
control_port = free_port()
control = frida.EndpointParameters(address="127.0.0.1", port=control_port, request_handler=Handler())
cluster = frida.EndpointParameters(address="127.0.0.1", port=free_port())
service = frida.PortalService(cluster, control)
service.start()
try:
with self.assertRaises(urllib.error.HTTPError) as raised:
urllib.request.urlopen(f"http://127.0.0.1:{control_port}/ping")
self.assertEqual(raised.exception.code, 500)
self.assertEqual(raised.exception.read(), b"boom")
finally:
service.stop()
@unittest.skipIf(sys.platform == "win32", "requires a POSIX shell")
def test_spawn_accepts_program_as_list(self):
import os
import tempfile
import time
device = self.frida.get_local_device()
fd, path = tempfile.mkstemp()
os.close(fd)
try:
pid = device.spawn(["/bin/sh", "-c", f'printf hi > "{path}"'])
device.resume(pid)
deadline = time.time() + 5
content = ""
while time.time() < deadline:
with open(path) as f:
content = f.read()
if content:
break
time.sleep(0.05)
finally:
os.unlink(path)
self.assertEqual(content, "hi")
@unittest.skipIf(sys.platform == "win32", "requires a POSIX shell")
def test_spawn_accepts_env_as_dict(self):
import os
import tempfile
import time
device = self.frida.get_local_device()
fd, path = tempfile.mkstemp()
os.close(fd)
try:
pid = device.spawn(
"/bin/sh",
argv=["/bin/sh", "-c", f'printf %s "$FRIDA_ENVP_TEST" > "{path}"'],
env={"FRIDA_ENVP_TEST": "envp-as-dict"},
)
device.resume(pid)
deadline = time.time() + 5
content = ""
while time.time() < deadline:
with open(path) as f:
content = f.read()
if content:
break
time.sleep(0.05)
finally:
os.unlink(path)
self.assertEqual(content, "envp-as-dict")
@unittest.skipIf(sys.platform == "win32", "requires a POSIX shell")
def test_spawn_accepts_envp_and_bytes_argv(self):
import os
import tempfile
import time
device = self.frida.get_local_device()
fd, path = tempfile.mkstemp()
os.close(fd)
try:
pid = device.spawn(
[b"/bin/sh", b"-c", f'printf %s "$FRIDA_ENVP2" > "{path}"'],
envp={"FRIDA_ENVP2": "envp-full"},
)
device.resume(pid)
deadline = time.time() + 5
content = ""
while time.time() < deadline:
with open(path) as f:
content = f.read()
if content:
break
time.sleep(0.05)
finally:
os.unlink(path)
self.assertEqual(content, "envp-full")
def test_toplevel_convenience_functions(self):
frida = self.frida
self.assertRegex(frida.__version__, r"\d+\.\d+")
for name in (
"spawn",
"resume",
"kill",
"attach",
"inject_library_file",
"inject_library_blob",
"enumerate_devices",
"get_device_matching",
"shutdown",
"query_system_parameters",
):
self.assertTrue(callable(getattr(frida, name)), name)
self.assertIn("local", [d.id for d in frida.enumerate_devices()])
self.assertEqual(frida.get_device_matching(lambda d: d.type == "local").id, "local")
@unittest.skipIf(sys.platform == "win32", "requires a POSIX shell")
def test_toplevel_attach_and_query(self):
frida = self.frida
pid = frida.spawn(["/bin/sh", "-c", "sleep 30"])
try:
session = frida.attach(pid)
session.detach()
finally:
frida.kill(pid)
params = frida.query_system_parameters()
self.assertIn("arch", params)
def test_device_type_property(self):
device = self.frida.get_local_device()
self.assertEqual(device.type, "local")
def test_bool_accessors_are_properties(self):
device = self.frida.get_local_device()
self.assertIs(device.is_lost, False)
session = device.attach(0)
self.assertIs(session.is_detached, False)
session.detach()
self.assertIs(session.is_detached, True)
def test_device_get_process_by_name(self):
device = self.frida.get_local_device()
with self.assertRaises(self.frida.ProcessNotFoundError):
device.get_process("frida-nonexistent-process-zzz")
@unittest.skipIf(sys.platform == "win32", "requires a POSIX shell")
def test_attach_accepts_pid(self):
device = self.frida.get_local_device()
pid = device.spawn(["/bin/sh", "-c", "sleep 30"])
try:
session = device.attach(pid)
session.detach()
finally:
device.kill(pid)
def test_facade_repr(self):
device = self.frida.get_local_device()
self.assertRegex(repr(device), r"Device\(id='local'")
process = self.frida.get_local_device().enumerate_processes()[0]
self.assertRegex(repr(process), r"Process\(pid=\d+")
def test_create_script_positional_name(self):
session = self.frida.get_local_device().attach(0)
try:
script = session.create_script("rpc.exports={ping:()=>1};", "my-script")
script.load()
self.assertEqual(script.exports_sync.ping(), 1)
finally:
session.detach()
def test_create_script_from_bytes_positional_name(self):
session = self.frida.get_local_device().attach(0)
try:
bytecode = session.compile_script("rpc.exports={ping:()=>1};", "n")
script = session.create_script_from_bytes(bytecode, "my-script")
script.load()
self.assertEqual(script.exports_sync.ping(), 1)
finally:
session.detach()
def test_file_monitor_reports_changes(self):
import os
import tempfile
import time
directory = tempfile.mkdtemp()
target = os.path.join(directory, "watched.txt")
monitor = self.frida.FileMonitor(target)
events = []
monitor.on("change", lambda *args: events.append(args))
monitor.enable()
try:
with open(target, "w") as f:
f.write("hi")
deadline = time.time() + 3
while not events and time.time() < deadline:
time.sleep(0.05)
finally:
monitor.disable()
self.assertTrue(events)
self.assertEqual(events[0][0], target)
def test_frida_core_alias(self):
import frida.core
self.assertIs(frida.core.Session, self.frida.Session)
self.assertIs(frida.core.Device, self.frida.Device)
def test_device_bus_post_accepts_dict(self):
bus = self.frida.get_local_device().bus
self.assertTrue(callable(bus.post))
bus.post({"type": "ping"})
def test_script_options_snapshot_setter(self):
options = self._new_script_options()
options.snapshot = b"\x00\x01\x02"
self.assertEqual(options.snapshot, b"\x00\x01\x02")
def _new_script_options(self):
from frida import _frida
return _frida.ScriptOptions()
def test_iostream_exposes_old_api(self):
for name in ("is_closed", "close", "read", "read_all", "write", "write_all"):
self.assertTrue(hasattr(self.frida.IOStream, name))
def test_get_local_device_returns_wrapped_device(self):
device = self.frida.get_local_device()
self.assertIsInstance(device, self.frida.Device)
self.assertEqual(device.id, "local")
def test_device_manager_is_a_singleton(self):
import frida._frida as _frida
import frida.aio
underlying = _frida.get_device_manager()
self.assertIs(_frida.get_device_manager(), underlying)
self.assertIs(self.frida.get_device_manager()._impl, underlying)
self.assertIs(frida.aio.get_device_manager()._impl, underlying)
def test_aio_get_local_device(self):
import asyncio
import frida.aio
device = asyncio.run(frida.aio.get_local_device())
self.assertIsInstance(device, frida.aio.Device)
self.assertEqual(device.id, "local")
def test_method_with_options_parameter_returns_object(self):
import os
device = self.frida.get_local_device()
process = device.get_process_by_pid(os.getpid())
self.assertIsInstance(process, self.frida.Process)
self.assertEqual(process.pid, os.getpid())
def test_method_with_options_parameter_maps_errors(self):
device = self.frida.get_local_device()
with self.assertRaises(self.frida.ProcessNotFoundError):
device.attach(999999)
def test_list_return_marshals_to_wrapped_objects(self):
devices = self.frida.get_device_manager().enumerate_devices()
self.assertIsInstance(devices, list)
self.assertTrue(all(isinstance(d, self.frida.Device) for d in devices))
self.assertIn("local", [d.id for d in devices])
def test_vardict_return_marshals_to_dict(self):
params = self.frida.get_local_device().query_system_parameters()
self.assertIsInstance(params, dict)
self.assertIn("arch", params)
self.assertIsInstance(params["os"], dict)
def test_signal_handler_is_invoked(self):
cancellable = self.frida.Cancellable()
fired = []
cancellable.on("cancelled", lambda: fired.append(True))
cancellable.cancel()
self.assertEqual(fired, [True])
def test_removed_signal_handler_is_not_invoked(self):
cancellable = self.frida.Cancellable()
hits = []
def handler():
hits.append(True)
cancellable.on("cancelled", handler)
cancellable.off("cancelled", handler)
cancellable.cancel()
self.assertEqual(hits, [])
def test_custom_base_type_still_exposes_signals(self):
device = self.frida.get_local_device()
callback = lambda *args: None
device.on("spawn-added", callback)
device.off("spawn-added", callback)
def test_unknown_signal_name_raises(self):
with self.assertRaises(ValueError):
self.frida.get_device_manager().on("nonexistent", lambda: None)
def test_signal_handler_can_receive_the_emitter(self):
cancellable = self.frida.Cancellable()
received = []
cancellable.on("cancelled", lambda sender: received.append(sender))
cancellable.cancel()
self.assertEqual(len(received), 1)
self.assertIsInstance(received[0], self.frida.Cancellable)
def test_signal_handler_with_too_many_arguments_raises(self):
cancellable = self.frida.Cancellable()
with self.assertRaises(TypeError):
cancellable.on("cancelled", lambda a, b, c: None)
def test_cancellable_is_cancelled_is_a_property(self):
cancellable = self.frida.Cancellable()
self.assertFalse(cancellable.is_cancelled)
cancellable.cancel()
self.assertTrue(cancellable.is_cancelled)
def test_cancellable_pollfd_round_trips(self):
cancellable = self.frida.Cancellable()
pollfd = cancellable.get_pollfd()
with pollfd as fd:
self.assertIsInstance(fd, int)
pollfd.release()
pollfd.release()
def test_rpc_round_trip(self):
session = self.frida.get_local_device().attach(0)
try:
script = session.create_script("rpc.exports = { add: (a, b) => a + b, echo: (x) => x };")
script.load()
self.assertEqual(sorted(dir(script.exports_sync)), ["add", "echo"])
self.assertEqual(script.exports_sync.add(3, 4), 7)
self.assertEqual(script.exports_sync.echo("hello"), "hello")
finally:
session.detach()
def test_script_rpc_error_propagates(self):
session = self.frida.get_local_device().attach(0)
try:
script = session.create_script("rpc.exports = { boom: () => { throw new Error('kaboom'); } };")
script.load()
with self.assertRaises(self.frida.RPCException):
script.exports_sync.boom()
finally:
session.detach()
def test_script_log_handler_receives_console_log(self):
session = self.frida.get_local_device().attach(0)
try:
script = session.create_script("console.log('hello from script');")
logs = []
script.set_log_handler(lambda level, text: logs.append((level, text)))
script.load()
self.assertIn(("info", "hello from script"), logs)
finally:
session.detach()
def test_script_list_exports(self):
session = self.frida.get_local_device().attach(0)
try:
script = session.create_script("rpc.exports = { a: () => 1, b: () => 2 };")
script.load()
self.assertEqual(sorted(script.list_exports_sync()), ["a", "b"])
finally:
session.detach()
def test_script_message_round_trip(self):
session = self.frida.get_local_device().attach(0)
try:
script = session.create_script("send({ hi: 1 });")
received = []
script.on("message", lambda message, data: received.append(message))
script.load()
self.assertEqual(received[0]["type"], "send")
self.assertEqual(received[0]["payload"], {"hi": 1})
finally:
session.detach()
def test_script_rpc_raises_after_destroy(self):
session = self.frida.get_local_device().attach(0)
try:
script = session.create_script("rpc.exports = { ping: () => 1 };")
script.load()
script.unload()
with self.assertRaises(self.frida.InvalidOperationError):
script.exports_sync.ping()
finally:
session.detach()
def test_rpc_exception_str_surfaces_message(self):
self.assertEqual(str(self.frida.RPCException("type", "name", "boom")), "boom")
self.assertEqual(str(self.frida.RPCException("boom")), "boom")
def test_sync_cancellable_context_cancels_blocking_call(self):
import threading
device = self.frida.get_local_device()
cancellable = self.frida.Cancellable()
threading.Timer(0.2, cancellable.cancel).start()
with self.assertRaises(self.frida.OperationCancelledError):
with cancellable:
device.get_process_by_name("frida-nonexistent-zzz", timeout=5000)
def test_aio_cancellable_context_cancels_awaitable(self):
import asyncio
import frida.aio
async def scenario():
device = await frida.aio.get_local_device()
cancellable = frida.aio.Cancellable()
async def cancel_soon():
await asyncio.sleep(0.2)
cancellable.cancel()
asyncio.ensure_future(cancel_soon())
with self.assertRaises(self.frida.OperationCancelledError):
with cancellable:
await device.get_process_by_name("frida-nonexistent-zzz", timeout=5000)
asyncio.run(scenario())
def test_attach_accepts_options_as_kwargs(self):
session = self.frida.get_local_device().attach(0, realm="native", persist_timeout=0)
session.detach()
def test_invalid_option_is_reported(self):
with self.assertRaises(AttributeError):
self.frida.get_local_device().attach(0, no_such_option=123)
def test_compile_script_returns_bytes(self):
session = self.frida.get_local_device().attach(0)
try:
data = session.compile_script("const x = 42;")
self.assertIsInstance(data, bytes)
self.assertGreater(len(data), 0)
finally:
session.detach()
def test_aio_rpc_round_trip(self):
import asyncio
import frida.aio
async def scenario():
device = await frida.aio.get_local_device()
session = await device.attach(0)
try:
script = await session.create_script("rpc.exports = { add: (a, b) => a + b };")
await script.load()
self.assertEqual(await script.exports.add(3, 4), 7)
finally:
await session.detach()
asyncio.run(scenario())
if __name__ == "__main__":
unittest.main()
tests/test_core.py
-49
View File
@@ -1,49 +0,0 @@
import threading
import time
import unittest
import frida
class TestCore(unittest.TestCase):
def test_enumerate_devices(self):
devices = frida.get_device_manager().enumerate_devices()
self.assertTrue(len(devices) > 0)
def test_get_existing_device(self):
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")
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.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.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
)
def cancel_after_100ms():
time.sleep(0.1)
cancellable.cancel()
threading.Thread(target=cancel_after_100ms).start()
self.assertRaisesRegex(frida.OperationCancelledError, "operation was cancelled", wait_for_nonexistent)
if __name__ == "__main__":
unittest.main()
tests/test_pep503_page_parser.py
View File
Whitespace-only changes.
tests/test_rpc.py
-171
View File
@@ -1,171 +0,0 @@
import subprocess
import threading
import time
import unittest
import frida
from .data import target_program
class TestRpc(unittest.TestCase):
target: subprocess.Popen
session: frida.core.Session
@classmethod
def setUp(cls):
cls.target = subprocess.Popen([target_program], stdin=subprocess.PIPE)
# TODO: improve injectors to handle injection into a process that hasn't yet finished initializing
time.sleep(0.05)
cls.session = frida.attach(cls.target.pid)
@classmethod
def tearDown(cls):
cls.session.detach()
cls.target.terminate()
cls.target.stdin.close()
cls.target.wait()
def test_basics(self):
script = self.session.create_script(
name="test-rpc",
source="""\
rpc.exports = {
add(a, b) {
const result = a + b;
if (result < 0)
throw new Error("No");
return result;
},
sub(a, b) {
return a - b;
},
speak() {
const buf = Memory.allocUtf8String("Yo");
return Memory.readByteArray(buf, 2);
},
speakWithMetadata() {
const buf = Memory.allocUtf8String("Yo");
return ['soft', Memory.readByteArray(buf, 2)];
},
processData(val, data) {
return { val, dump: hexdump(data, { header: false }) };
},
};
""",
)
script.load()
agent = script.exports_sync
self.assertEqual(agent.add(2, 3), 5)
self.assertEqual(agent.sub(5, 3), 2)
self.assertRaises(Exception, lambda: agent.add(1, -2))
self.assertEqual(agent.speak(), b"\x59\x6f")
meta, data = agent.speak_with_metadata()
self.assertEqual(meta, "soft")
self.assertEqual(data, b"\x59\x6f")
result = agent.process_data(1337, b"\x13\x37")
self.assertEqual(result["val"], 1337)
self.assertEqual(result["dump"], "00000000 13 37 .7")
def test_post_failure(self):
script = self.session.create_script(
name="test-rpc",
source="""\
rpc.exports = {
init: function () {
},
};
""",
)
script.load()
agent = script.exports_sync
self.session.detach()
self.assertRaisesScriptDestroyed(lambda: agent.init())
self.assertEqual(script._pending, {})
def test_unload_mid_request(self):
script = self.session.create_script(
name="test-rpc",
source="""\
rpc.exports = {
waitForever: function () {
return new Promise(function () {});
},
};
""",
)
script.load()
agent = script.exports_sync
def unload_script_after_100ms():
time.sleep(0.1)
script.unload()
threading.Thread(target=unload_script_after_100ms).start()
self.assertRaisesScriptDestroyed(lambda: agent.wait_forever())
self.assertEqual(script._pending, {})
def test_detach_mid_request(self):
script = self.session.create_script(
name="test-rpc",
source="""\
rpc.exports = {
waitForever: function () {
return new Promise(function () {});
},
};
""",
)
script.load()
agent = script.exports_sync
def terminate_target_after_100ms():
time.sleep(0.1)
self.target.terminate()
threading.Thread(target=terminate_target_after_100ms).start()
self.assertRaisesScriptDestroyed(lambda: agent.wait_forever())
self.assertEqual(script._pending, {})
def test_cancellation_mid_request(self):
script = self.session.create_script(
name="test-rpc",
source="""\
rpc.exports = {
waitForever: function () {
return new Promise(function () {});
},
};
""",
)
script.load()
agent = script.exports_sync
def cancel_after_100ms():
time.sleep(0.1)
cancellable.cancel()
cancellable = frida.Cancellable()
threading.Thread(target=cancel_after_100ms).start()
self.assertRaisesOperationCancelled(lambda: agent.wait_forever(cancellable=cancellable))
self.assertEqual(script._pending, {})
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.assertRaisesRegex(frida.InvalidOperationError, "script has been destroyed", operation)
def assertRaisesOperationCancelled(self, operation):
self.assertRaisesRegex(frida.OperationCancelledError, "operation was cancelled", operation)
if __name__ == "__main__":
unittest.main()