Add print_alive_tasks() debug binding

Expose g_task_print_alive_tasks() as _frida.print_alive_tasks(), wrapped as frida.print_alive_tasks() in both facades. Requires a GLib built with G_ENABLE_DEBUG.
subprojects: Prepare for release
2026-11-11 17:12:19 -10:00 · 2026-07-28 16:42:54 +02:00 · 2026-07-22 00:30:14 +02:00 · 2026-07-22 00:29:03 +02:00 · 2026-07-22 00:23:55 +02:00 · 2026-07-22 00:23:55 +02:00
73 changed files with 7234 additions and 8460 deletions
@@ -4,13 +4,12 @@ jobs:
  black:
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v7
      - uses: psf/black@stable
  isort:
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v7
-      - uses: actions/setup-python@v4
+      - uses: actions/setup-python@v6
-        with:
+      - run: pip install isort
-          python-version: 3.8
+      - run: isort --check-only --diff .
      - uses: jamescurtin/isort-action@master
@@ -4,12 +4,13 @@ jobs:
  pyflakes:
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v7
-      - 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: actions/checkout@v7
      - uses: jpetrucciani/mypy-check@master
        with:
-          mypy_flags: '--exclude examples --exclude setup'
+          mypy_flags: '--exclude examples --exclude setup --exclude frida/frida_bindgen/assets'
@@ -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
@@ -1,5 +1,7 @@
 *.pyc
 *.abi3.so
 /*.egg-info/
 /.cache/
 /build/
 /deps/
 /dist/
@@ -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
@@ -1,8 +1,8 @@
 include meson.build
 include frida/meson.build
-include frida/_frida/extension.c
+include frida/extension.version
-include frida/_frida/extension.version
+recursive-include frida/frida_bindgen *.py *.c *.h
-include frida/_frida/meson.build
+recursive-include frida-bindgen *.py
 include subprojects/frida-core.wrap
 include BSDmakefile
@@ -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
@@ -42,18 +42,16 @@ class Application:
        print("✔ enable_child_gating()")
        session.enable_child_gating()
        print("✔ create_script()")
-        script = session.create_script(
+        script = session.create_script("""\
-            """\
+Interceptor.attach(Module.getGlobalExportByName('open'), {
-Interceptor.attach(Module.getExportByName(null, 'open'), {
+  onEnter(args) {
  onEnter: function (args) {
    send({
      type: 'open',
-      path: Memory.readUtf8String(args[0])
+      path: args[0].readUtf8String()
    });
  }
 });
-"""
+""")
        )
        script.on("message", lambda message, data: self._reactor.schedule(lambda: self._on_message(pid, message)))
        print("✔ load()")
        script.load()
@@ -15,7 +15,7 @@ def on_uninjected(id):
    print("on_uninjected id=%u" % id)
-(target, library_path) = sys.argv[1:]
+target, library_path = sys.argv[1:]
 device = frida.get_local_device()
 device.on("uninjected", on_uninjected)
@@ -15,7 +15,7 @@ def on_uninjected(id):
    print("on_uninjected id=%u" % id)
-(target, library_path) = sys.argv[1:]
+target, library_path = sys.argv[1:]
 device = frida.get_local_device()
 device.on("uninjected", on_uninjected)
@@ -1,8 +1,7 @@
 import frida
 session = frida.attach("Twitter")
-script = session.create_script(
+script = session.create_script("""\
    """\
 rpc.exports = {
  hello: function () {
    return 'Hello';
@@ -11,8 +10,7 @@ rpc.exports = {
    oops;
  }
 };
-"""
+""")
 )
 script.load()
 api = script.exports_sync
 print("api.hello() =>", api.hello())
@@ -22,8 +22,7 @@ class Application:
        self._session = session
        session.on("detached", lambda *args: self._reactor.schedule(lambda: self._on_detached(*args)))
-        script = session.create_script(
+        script = session.create_script("""
            """
 let _puts = null;
 Interceptor.attach(DebugSymbol.getFunctionByName('f'), {
@@ -48,8 +47,7 @@ function puts(s) {
  }
  _puts(Memory.allocUtf8String(s));
 }
-"""
+""")
        )
        self._script = script
        script.on("message", lambda *args: self._reactor.schedule(lambda: self._on_message(*args)))
        script.load()
@@ -0,0 +1,78 @@
 import threading
 from frida_tools.application import Reactor
 import frida
 class Application:
    def __init__(self):
        self._stop_requested = threading.Event()
        self._reactor = Reactor(run_until_return=lambda reactor: self._stop_requested.wait())
        self._device = frida.get_usb_device()
        self._sessions = set()
        self._sessions_lock = threading.Lock()
        self._device.on("spawn-added", lambda spawn: self._reactor.schedule(lambda: self._on_spawn_added(spawn)))
        self._device.on("spawn-removed", lambda spawn: self._reactor.schedule(lambda: self._on_spawn_removed(spawn)))
    def run(self):
        self._reactor.schedule(lambda: self._start())
        self._reactor.run()
    def _start(self):
        self._device.enable_spawn_gating()
    def _instrument(self, pid):
        print(f"✔ attach(pid={pid})")
        session = self._device.attach(pid)
        session.on("detached", lambda reason: self._reactor.schedule(lambda: self._on_detached(pid, session, reason)))
        print("✔ create_script()")
        script = session.create_script("""\
 const puts = new NativeFunction(Module.getGlobalExportByName('puts'), 'int', ['pointer']);
 puts(Memory.allocUtf8String('Hello from Frida agent'));
 Interceptor.attach(Module.getGlobalExportByName('open'), {
  onEnter(args) {
    send({
      type: 'open',
      path: args[0].readUtf8String()
    });
  }
 });
 """)
        script.on("message", lambda message, data: self._reactor.schedule(lambda: self._on_message(pid, message)))
        print("✔ load()")
        script.load()
        print(f"✔ resume(pid={pid})")
        self._device.resume(pid)
        with self._sessions_lock:
            self._sessions.add(session)
    def _on_spawn_added(self, spawn):
        print(f"⚡ spawn_added: {spawn}")
        t = threading.Thread(target=self._handle_spawn, args=(spawn,))
        t.start()
    def _handle_spawn(self, spawn):
        if "/bin/ls" in spawn.identifier:
            self._instrument(spawn.pid)
        else:
            pid = spawn.pid
            print(f"✔ resume(pid={pid})")
            self._device.resume(pid)
    def _on_spawn_removed(self, spawn):
        print(f"⚡ spawn_removed: {spawn}")
    def _on_detached(self, pid, session, reason):
        print(f"⚡ detached: pid={pid}, reason='{reason}'")
        with self._sessions_lock:
            self._sessions.remove(session)
    def _on_message(self, pid, message):
        print(f"⚡ message: pid={pid}, payload={message['payload']}")
 app = Application()
 app.run()
@@ -1,172 +0,0 @@
 from typing import Any, Callable, Dict, List, Optional, 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
 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
 ) -> core.Session:
    """
    Attach to a process
    :param target: the PID or name of the process
    """
    return get_local_device().attach(target, realm=realm, persist_timeout=persist_timeout)
 def inject_library_file(target: 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()
@@ -1,777 +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) -> "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 query_system_parameters(self) -> Dict[str, Any]:
        """
        Returns a dictionary of information about the host system.
        """
        ...
    def resume(self, pid: int) -> None:
        """
        Resume a process from the attachable state.
        """
        ...
    def spawn(
        self,
        program: str,
        argv: Union[None, List[Union[str, bytes]], Tuple[Union[str, bytes]]] = None,
        envp: Optional[Dict[str, str]] = None,
        env: Optional[Dict[str, str]] = None,
        cwd: Optional[str] = None,
        stdio: Optional[str] = None,
        **kwargs: Any,
    ) -> int:
        """
        Spawn a process into an attachable state.
        """
        ...
 class DeviceManager(Object):
    def add_remote_device(
        self,
        address: str,
        certificate: Optional[str] = None,
        origin: Optional[str] = None,
        token: Optional[str] = None,
        keepalive_interval: Optional[int] = None,
    ) -> Device:
        """
        Add a remote device.
        """
        ...
    def close(self) -> None:
        """
        Close the device manager.
        """
        ...
    def enumerate_devices(self) -> List[Device]:
        """
        Enumerate devices.
        """
        ...
    def get_device_matching(self, predicate: Callable[[Device], bool], timeout: int) -> Device:
        """
        Get device matching predicate.
        """
        ...
    def remove_remote_device(self, address: str) -> None:
        """
        Remove a remote device.
        """
        ...
 class EndpointParameters(Object): ...
 class FileMonitor(Object):
    def disable(self) -> None:
        """
        Disable the file monitor.
        """
        ...
    def enable(self) -> None:
        """
        Enable the file monitor.
        """
        ...
 class IOStream(Object):
    def close(self) -> None:
        """
        Close the stream.
        """
        ...
    def is_closed(self) -> bool:
        """
        Query whether the stream is closed.
        """
        ...
    def read(self, size: int) -> bytes:
        """
        Read up to the specified number of bytes from the stream.
        """
        ...
    def read_all(self, size: int) -> bytes:
        """
        Read exactly the specified number of bytes from the stream.
        """
        ...
    def write(self, data: bytes) -> int:
        """
        Write as much as possible of the provided data to the stream.
        """
        ...
    def write_all(self, data: bytes) -> None:
        """
        Write all of the provided data to the stream.
        """
        ...
 class PortalMembership(Object):
    def terminate(self) -> None:
        """
        Terminate the membership.
        """
        ...
 class PortalService(Object):
    @property
    def device(self) -> Device:
        """
        Device for in-process control.
        """
        ...
    def broadcast(self, message: str, data: Optional[Union[str, bytes]] = None) -> None:
        """
        Broadcast a message to all control channels.
        """
        ...
    def enumerate_tags(self, connection_id: int) -> List[str]:
        """
        Enumerate tags of a specific connection.
        """
        ...
    def kick(self, connection_id: int) -> None:
        """
        Kick out a specific connection.
        """
        ...
    def narrowcast(self, tag: str, message: str, data: Optional[Union[str, bytes]] = None) -> None:
        """
        Post a message to control channels with a specific tag.
        """
        ...
    def post(self, connection_id: int, message: str, data: Optional[Union[str, bytes]] = None) -> None:
        """
        Post a message to a specific control channel.
        """
        ...
    def start(self) -> None:
        """
        Start listening for incoming connections.
        """
        ...
    def stop(self) -> None:
        """
        Stop listening for incoming connections, and kick any connected clients.
        """
        ...
    def tag(self, connection_id: int, tag: str) -> None:
        """
        Tag a specific control channel.
        """
        ...
    def untag(self, connection_id: int, tag: str) -> None:
        """
        Untag a specific control channel.
        """
        ...
 class Process(Object):
    @property
    def pid(self) -> int:
        """
        Process ID.
        """
        ...
    @property
    def name(self) -> str:
        """
        Human-readable process name.
        """
        ...
    @property
    def parameters(self) -> Dict[str, Any]:
        """
        Parameters.
        """
        ...
 class Relay(Object):
    def __init__(self, address: str, username: str, password: str, kind: str) -> None: ...
    @property
    def address(self) -> str:
        """
        Network address or address:port of the TURN server.
        """
        ...
    @property
    def kind(self) -> str:
        """
        Relay kind. One of: turn-udp, turn-tcp, turn-tls.
        """
        ...
    @property
    def password(self) -> str:
        """
        The TURN password to use for the allocate request.
        """
        ...
    @property
    def username(self) -> str:
        """
        The TURN username to use for the allocate request.
        """
        ...
 class Script(Object):
    def eternalize(self) -> None:
        """
        Eternalize the script.
        """
        ...
    def is_destroyed(self) -> bool:
        """
        Query whether the script has been destroyed.
        """
        ...
    def load(self) -> None:
        """
        Load the script.
        """
        ...
    def post(self, message: str, data: Optional[Union[str, bytes]] = None) -> None:
        """
        Post a JSON-encoded message to the script.
        """
        ...
    def unload(self) -> None:
        """
        Unload the script.
        """
        ...
    def enable_debugger(self, port: Optional[int]) -> None:
        """
        Enable the Node.js compatible script debugger
        """
        ...
    def disable_debugger(self) -> None:
        """
        Disable the Node.js compatible script debugger
        """
        ...
 class 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,
        source_maps: Optional[str] = None,
        compression: Optional[str] = None,
    ) -> str:
        """
        Build an agent.
        """
        ...
    def watch(
        self,
        entrypoint: str,
        project_root: Optional[str] = None,
        source_maps: Optional[str] = None,
        compression: Optional[str] = None,
    ) -> None:
        """
        Continuously build an agent.
        """
        ...
 __version__: str
@@ -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',
 )
@@ -0,0 +1,4 @@
 from .cli import main
 if __name__ == "__main__":
    main()
@@ -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;
 }
@@ -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;
@@ -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);
@@ -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
@@ -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;
 };
@@ -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);
@@ -0,0 +1,23 @@
 def _setup(self) -> None:
    self._message_handlers: List[BusMessageCallback] = []
    self._impl.on("message", self._on_message)
 def on(self, signal: str, callback: Callable[..., Any]) -> None:
    if signal == "message":
        self._message_handlers.append(callback)
    else:
        self._impl.on(signal, _make_signal_handler(callback))
 def off(self, signal: str, callback: Callable[..., Any]) -> None:
    if signal == "message":
        self._message_handlers.remove(callback)
    else:
        self._impl.off(signal, callback)
 def _on_message(self, raw_message: str, data: Optional[bytes]) -> None:
    message = json.loads(raw_message)
    for handler in list(self._message_handlers):
        try:
            handler(message, data)
        except Exception:
            traceback.print_exc()
@@ -0,0 +1,27 @@
 def _setup(self) -> None:
    self._message_handlers: List[BusMessageCallback] = []
    self._loop = asyncio.get_event_loop()
    self._impl.on("message", self._on_message)
 def on(self, signal: str, callback: Callable[..., Any]) -> None:
    if signal == "message":
        self._message_handlers.append(callback)
    else:
        self._impl.on(signal, _make_signal_handler(callback))
 def off(self, signal: str, callback: Callable[..., Any]) -> None:
    if signal == "message":
        self._message_handlers.remove(callback)
    else:
        self._impl.off(signal, callback)
 def _on_message(self, raw_message: str, data: Optional[bytes]) -> None:
    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: BusMessageCallback, message: Mapping[Any, Any], data: Optional[bytes]) -> None:
    try:
        handler(message, data)
    except Exception:
        traceback.print_exc()
@@ -0,0 +1,49 @@
 def __enter__(self) -> "Cancellable":
    stack = getattr(_current_cancellable, "stack", None)
    if stack is None:
        stack = []
        _current_cancellable.stack = stack
    stack.append(self)
    return self
 def __exit__(self, *exc: Any) -> Literal[False]:
    _current_cancellable.stack.pop()
    return False
@classmethod
 def get_current(cls) -> "Cancellable":
    return _current_cancellable_get()
 def raise_if_cancelled(self) -> None:
    if self._impl.is_cancelled():
        raise _frida.OperationCancelledError("operation was cancelled")
 def get_pollfd(self) -> "CancellablePollFD":
    return CancellablePollFD(self._impl)
 def _setup(self) -> None:
    self._cancel_handlers: Dict[int, Callable[..., Any]] = {}
    self._next_cancel_handler_id = 1
 def connect(self, callback: Callable[..., Any]) -> int:
    if self._impl.is_cancelled():
        callback()
        return 0
    handler_id = self._next_cancel_handler_id
    self._next_cancel_handler_id = handler_id + 1
    def handler(*args: Any) -> None:
        callback()
    self._cancel_handlers[handler_id] = handler
    self._impl.on("cancelled", handler)
    return handler_id
 def disconnect(self, handler_id: int) -> None:
    handler = self._cancel_handlers.pop(handler_id, None)
    if handler is not None:
        self._impl.off("cancelled", handler)
@@ -0,0 +1,45 @@
 def __enter__(self) -> "Cancellable":
    self._token = _current_cancellable.set(self)
    return self
 def __exit__(self, *exc: Any) -> Literal[False]:
    _current_cancellable.reset(self._token)
    return False
@classmethod
 def get_current(cls) -> "Cancellable":
    return _current_cancellable.get()
 def raise_if_cancelled(self) -> None:
    if self._impl.is_cancelled():
        raise _frida.OperationCancelledError("operation was cancelled")
 def get_pollfd(self) -> "CancellablePollFD":
    return CancellablePollFD(self._impl)
 def _setup(self) -> None:
    self._cancel_handlers: Dict[int, Callable[..., Any]] = {}
    self._next_cancel_handler_id = 1
 def connect(self, callback: Callable[..., Any]) -> int:
    if self._impl.is_cancelled():
        callback()
        return 0
    handler_id = self._next_cancel_handler_id
    self._next_cancel_handler_id = handler_id + 1
    def handler(*args: Any) -> None:
        callback()
    self._cancel_handlers[handler_id] = handler
    self._impl.on("cancelled", handler)
    return handler_id
 def disconnect(self, handler_id: int) -> None:
    handler = self._cancel_handlers.pop(handler_id, None)
    if handler is not None:
        self._impl.off("cancelled", handler)
@@ -0,0 +1,40 @@
 class CancellablePollFD:
    def __init__(self, cancellable: _frida.Cancellable) -> None:
        self.handle = cancellable.get_fd()
        self._cancellable: Optional[_frida.Cancellable] = cancellable
    def __del__(self) -> None:
        self.release()
    def release(self) -> None:
        if self._cancellable is not None:
            if self.handle != -1:
                self._cancellable.release_fd()
                self.handle = -1
            self._cancellable = None
    def __repr__(self) -> str:
        return repr(self.handle)
    def __enter__(self) -> int:
        return self.handle
    def __exit__(self, *exc: Any) -> None:
        self.release()
 P = ParamSpec("P")
 R = TypeVar("R", covariant=True)
 def cancellable(f: Callable[P, R]) -> Callable[P, R]:
    @functools.wraps(f)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        cancellable = kwargs.pop("cancellable", None)
        if cancellable is not None:
            with cast(Cancellable, cancellable):
                return f(*args, **kwargs)
        return f(*args, **kwargs)
    return wrapper
@@ -0,0 +1,40 @@
 class CancellablePollFD:
    def __init__(self, cancellable: _frida.Cancellable) -> None:
        self.handle = cancellable.get_fd()
        self._cancellable: Optional[_frida.Cancellable] = cancellable
    def __del__(self) -> None:
        self.release()
    def release(self) -> None:
        if self._cancellable is not None:
            if self.handle != -1:
                self._cancellable.release_fd()
                self.handle = -1
            self._cancellable = None
    def __repr__(self) -> str:
        return repr(self.handle)
    def __enter__(self) -> int:
        return self.handle
    def __exit__(self, *exc: Any) -> None:
        self.release()
 P = ParamSpec("P")
 R = TypeVar("R", covariant=True)
 def cancellable(f: Callable[P, R]) -> Callable[P, R]:
    @functools.wraps(f)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        cancellable = kwargs.pop("cancellable", None)
        if cancellable is not None:
            with cast(Cancellable, cancellable):
                return f(*args, **kwargs)
        return f(*args, **kwargs)
    return wrapper
@@ -0,0 +1,25 @@
 def get_device_matching(self, predicate: Callable[["Device"], bool], timeout: Union[int, float] = 0) -> "Device":
    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)
 def get_device(self, id: Optional[str], timeout: Union[int, float] = 0) -> "Device":
    return self.get_device_matching(lambda d: d.id == id, timeout)
 def get_local_device(self) -> "Device":
    return self.get_device_matching(lambda d: d.type == "local", 0)
 def get_remote_device(self) -> "Device":
    return self.get_device_matching(lambda d: d.type == "remote", 0)
 def get_usb_device(self, timeout: Union[int, float] = 0) -> "Device":
    return self.get_device_matching(lambda d: d.type == "usb", timeout)
@@ -0,0 +1,25 @@
 async def get_device_matching(self, predicate: Callable[["Device"], bool], timeout: Union[int, float] = 0) -> "Device":
    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)
 async def get_device(self, id: Optional[str], timeout: Union[int, float] = 0) -> "Device":
    return await self.get_device_matching(lambda d: d.id == id, timeout)
 async def get_local_device(self) -> "Device":
    return await self.get_device_matching(lambda d: d.type == "local", 0)
 async def get_remote_device(self) -> "Device":
    return await self.get_device_matching(lambda d: d.type == "remote", 0)
 async def get_usb_device(self, timeout: Union[int, float] = 0) -> "Device":
    return await self.get_device_matching(lambda d: d.type == "usb", timeout)
@@ -0,0 +1,25 @@
@property
 def type(self) -> str:
    return self.dtype
 def get_bus(self) -> "Bus":
    return self.bus
 def get_process(self, process_name: str) -> "Process":
    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: ProcessTarget) -> int:
    if isinstance(target, str):
        return self.get_process(target).pid
    return target
@@ -0,0 +1,25 @@
@property
 def type(self) -> str:
    return self.dtype
 def get_bus(self) -> "Bus":
    return self.bus
 async def get_process(self, process_name: str) -> "Process":
    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: ProcessTarget) -> int:
    if isinstance(target, str):
        return (await self.get_process(target)).pid
    return target
@@ -0,0 +1 @@
 ScriptExportsAsync = aio._ScriptExports
@@ -0,0 +1,12 @@
 class _Implementation:
    def _frida_dispatch(self, name: str, args: List[Any], completion: Any) -> None:
        def run() -> None:
            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()
@@ -0,0 +1,19 @@
 class _Implementation:
    _loop: Any
    def _frida_dispatch(self, name: str, args: List[Any], completion: Any) -> None:
        loop = self._loop
        async def run() -> None:
            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()))
@@ -0,0 +1,25 @@
 def close(self) -> None:
    _invoke(self._impl.close_async, (0,))
 def read(self, count: int) -> bytes:
    return _invoke(self._impl.input_stream.read_bytes_async, (count, 0))
 def read_all(self, count: int) -> bytes:
    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: bytes) -> int:
    return _invoke(self._impl.output_stream.write_bytes_async, (data, 0))
 def write_all(self, data: bytes) -> None:
    view = memoryview(data)
    while len(view) != 0:
        written = _invoke(self._impl.output_stream.write_bytes_async, (bytes(view), 0))
        view = view[written:]
@@ -0,0 +1,25 @@
 async def close(self) -> None:
    await _invoke(self._impl.close_async, (0,))
 async def read(self, count: int) -> bytes:
    return await _invoke(self._impl.input_stream.read_bytes_async, (count, 0))
 async def read_all(self, count: int) -> bytes:
    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: bytes) -> int:
    return await _invoke(self._impl.output_stream.write_bytes_async, (data, 0))
 async def write_all(self, data: bytes) -> None:
    view = memoryview(data)
    while len(view) != 0:
        written = await _invoke(self._impl.output_stream.write_bytes_async, (bytes(view), 0))
        view = view[written:]
@@ -0,0 +1,37 @@
 def _setup(self) -> None:
    self._authenticated_handlers: List[PortalServiceAuthenticatedCallback] = []
    self._message_handlers: List[PortalServiceMessageCallback] = []
    self._impl.on("authenticated", self._on_authenticated)
    self._impl.on("message", self._on_message)
 def on(self, signal: str, callback: Callable[..., Any]) -> None:
    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: str, callback: Callable[..., Any]) -> None:
    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: int, raw_session_info: str) -> None:
    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: int, raw_message: str, data: Optional[bytes]) -> None:
    message = json.loads(raw_message)
    for handler in list(self._message_handlers):
        try:
            handler(connection_id, message, data)
        except Exception:
            traceback.print_exc()
@@ -0,0 +1,38 @@
 def _setup(self) -> None:
    self._authenticated_handlers: List[PortalServiceAuthenticatedCallback] = []
    self._message_handlers: List[PortalServiceMessageCallback] = []
    self._loop = asyncio.get_event_loop()
    self._impl.on("authenticated", self._on_authenticated)
    self._impl.on("message", self._on_message)
 def on(self, signal: str, callback: Callable[..., Any]) -> None:
    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: str, callback: Callable[..., Any]) -> None:
    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: int, raw_session_info: str) -> None:
    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: int, raw_message: str, data: Optional[bytes]) -> None:
    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: Callable[..., Any], *args: Any) -> None:
    try:
        handler(*args)
    except Exception:
        traceback.print_exc()
@@ -0,0 +1,59 @@
 def _to_camel_case(name: str) -> str:
    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) -> str:
        return str(self.args[2]) if len(self.args) >= 3 else str(self.args[0])
 class _ScriptExports:
    def __init__(self, script: "Script") -> None:
        self._script = script
    def __getattr__(self, name: str) -> Callable[..., Any]:
        script = self._script
        js_name = _to_camel_case(name)
        def method(*args: Any) -> Any:
            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) -> List[str]:
        return self._script._list_exports()
 ScriptExportsSync = _ScriptExports
 ScriptExports = ScriptExportsSync
@dataclasses.dataclass
 class RPCResult:
    finished: bool = False
    value: Any = None
    error: Optional[Exception] = None
 def make_rpc_call_request(js_name: str, args: Sequence[Any]) -> Tuple[List[Any], Optional[bytes]]:
    if args and isinstance(args[-1], bytes):
        raw_args = args[:-1]
        data = args[-1]
    else:
        raw_args = args
        data = None
    return (["call", js_name, raw_args], data)
@@ -0,0 +1,55 @@
 def _to_camel_case(name: str) -> str:
    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) -> str:
        return str(self.args[2]) if len(self.args) >= 3 else str(self.args[0])
 class _ScriptExports:
    def __init__(self, script: "Script") -> None:
        self._script = script
    def __getattr__(self, name: str) -> Callable[..., Any]:
        script = self._script
        js_name = _to_camel_case(name)
        def method(*args: Any) -> Any:
            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
 ScriptExportsAsync = _ScriptExports
@dataclasses.dataclass
 class RPCResult:
    finished: bool = False
    value: Any = None
    error: Optional[Exception] = None
 def make_rpc_call_request(js_name: str, args: Sequence[Any]) -> Tuple[List[Any], Optional[bytes]]:
    if args and isinstance(args[-1], bytes):
        raw_args = args[:-1]
        data = args[-1]
    else:
        raw_args = args
        data = None
    return (["call", js_name, raw_args], data)
@@ -0,0 +1,111 @@
 def _setup(self) -> None:
    self.exports_sync = _ScriptExports(self)
    self._message_handlers: List[ScriptMessageCallback] = []
    self._log_handler: Callable[[str, str], None] = self.default_log_handler
    self._pending: Dict[int, Callable[[Any, Optional[Exception]], None]] = {}
    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) -> "_ScriptExports":
    return self.exports_sync
 def on(self, signal: str, callback: Callable[..., Any]) -> None:
    if signal == "message":
        self._message_handlers.append(callback)
    else:
        self._impl.on(signal, _make_signal_handler(callback))
 def off(self, signal: str, callback: Callable[..., Any]) -> None:
    if signal == "message":
        self._message_handlers.remove(callback)
    else:
        self._impl.off(signal, callback)
 def get_log_handler(self) -> Callable[[str, str], None]:
    return self._log_handler
 def set_log_handler(self, handler: Callable[[str, str], None]) -> None:
    self._log_handler = handler
 def default_log_handler(self, level: str, text: str) -> None:
    if level == "info":
        print(text, file=sys.stdout)
    else:
        print(text, file=sys.stderr)
 def list_exports_sync(self) -> List[str]:
    return self._rpc_request(["list"])
 def list_exports(self) -> List[str]:
    return self.list_exports_sync()
 def _list_exports(self) -> List[str]:
    return self.list_exports_sync()
 def _rpc_request(self, args: Any, data: Optional[bytes] = None) -> Any:
    outcome: Dict[str, Any] = {}
    cond = self._cond
    def complete(value: Any, error: Optional[Exception]) -> None:
        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: str, data: Optional[bytes]) -> None:
    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) -> None:
    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: int, operation: str, params: List[Any], data: Optional[Any]) -> None:
    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)
@@ -0,0 +1,98 @@
 def _setup(self) -> None:
    self.exports = _ScriptExports(self)
    self._message_handlers: List[ScriptMessageCallback] = []
    self._log_handler: Callable[[str, str], None] = self.default_log_handler
    self._pending: Dict[int, Callable[[Any, Optional[Exception]], None]] = {}
    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: str, callback: Callable[..., Any]) -> None:
    if signal == "message":
        self._message_handlers.append(callback)
    else:
        self._impl.on(signal, _make_signal_handler(callback))
 def off(self, signal: str, callback: Callable[..., Any]) -> None:
    if signal == "message":
        self._message_handlers.remove(callback)
    else:
        self._impl.off(signal, callback)
 def get_log_handler(self) -> Callable[[str, str], None]:
    return self._log_handler
 def set_log_handler(self, handler: Callable[[str, str], None]) -> None:
    self._log_handler = handler
 def default_log_handler(self, level: str, text: str) -> None:
    if level == "info":
        print(text, file=sys.stdout)
    else:
        print(text, file=sys.stderr)
 async def list_exports(self) -> List[str]:
    return await self._rpc_request(["list"])
 def _rpc_request(self, args: Any, data: Optional[bytes] = None) -> "asyncio.Future[Any]":
    loop = asyncio.get_running_loop()
    future: "asyncio.Future[Any]" = loop.create_future()
    def complete(value: Any, error: Optional[Exception]) -> None:
        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: str, data: Optional[bytes]) -> None:
    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: ScriptMessageCallback, message: ScriptMessage, data: Optional[bytes]) -> None:
    try:
        handler(message, data)
    except Exception:
        traceback.print_exc()
 def _on_destroyed(self) -> None:
    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: int, operation: str, params: List[Any], data: Optional[Any]) -> None:
    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)
@@ -0,0 +1,84 @@
 __version__ = _frida.__version__
 def get_device_manager() -> "DeviceManager":
    return _wrap(_frida.get_device_manager())
 def get_device(id: str, timeout: Union[int, float] = 0) -> "Device":
    return get_device_manager().get_device_by_id(id, int(timeout * 1000))
 def get_device_matching(predicate: Callable[["Device"], bool], timeout: Union[int, float] = 0) -> "Device":
    return get_device_manager().get_device_matching(predicate, timeout)
 def get_local_device() -> "Device":
    return get_device_manager().get_device_by_type("local", 0)
 def get_remote_device() -> "Device":
    return get_device_manager().get_device_by_type("remote", 0)
 def get_usb_device(timeout: Union[int, float] = 0) -> "Device":
    return get_device_manager().get_device_by_type("usb", int(timeout * 1000))
 def enumerate_devices() -> List["Device"]:
    return get_device_manager().enumerate_devices()
 def query_system_parameters() -> Dict[str, Any]:
    return get_local_device().query_system_parameters()
 def spawn(
    program: Union[str, List[Union[str, bytes]]],
    argv: Optional[List[Union[str, bytes]]] = None,
    envp: Optional[Dict[str, str]] = None,
    env: Optional[Dict[str, str]] = None,
    cwd: Optional[str] = None,
    stdio: Optional[str] = None,
    **aux: Any,
 ) -> int:
    return get_local_device().spawn(
        program, argv=argv, envp=envp, env=env, cwd=cwd, stdio=stdio, **aux
    )
 def resume(target: ProcessTarget) -> None:
    return get_local_device().resume(target)
 def kill(target: ProcessTarget) -> None:
    return get_local_device().kill(target)
 def attach(target: ProcessTarget, **kwargs: Any) -> "Session":
    return get_local_device().attach(target, **kwargs)
 def inject_library_file(target: ProcessTarget, path: str, entrypoint: str, data: str) -> int:
    return get_local_device().inject_library_file(target, path, entrypoint, data)
 def inject_library_blob(target: ProcessTarget, blob: bytes, entrypoint: str, data: str) -> int:
    return get_local_device().inject_library_blob(target, blob, entrypoint, data)
 def shutdown() -> None:
    get_device_manager().close()
 def make_auth_callback(callback: Callable[[str], Any]) -> Callable[[Any], str]:
    def authenticate(token: str) -> str:
        session_info = callback(token)
        return json.dumps(session_info)
    return authenticate
 # TEMPORARY DEBUG HACK
 def print_alive_tasks() -> None:
    _frida.print_alive_tasks()
@@ -0,0 +1,85 @@
 __version__ = _frida.__version__
 def get_device_manager() -> "DeviceManager":
    return _wrap(_frida.get_device_manager())
 async def get_device(id: str, timeout: Union[int, float] = 0) -> "Device":
    return await get_device_manager().get_device_by_id(id, int(timeout * 1000))
 async def get_device_matching(predicate: Callable[["Device"], bool], timeout: Union[int, float] = 0) -> "Device":
    return await get_device_manager().get_device_matching(predicate, timeout)
 async def get_local_device() -> "Device":
    return await get_device_manager().get_device_by_type("local", 0)
 async def get_remote_device() -> "Device":
    return await get_device_manager().get_device_by_type("remote", 0)
 async def get_usb_device(timeout: Union[int, float] = 0) -> "Device":
    return await get_device_manager().get_device_by_type("usb", int(timeout * 1000))
 async def enumerate_devices() -> List["Device"]:
    return await get_device_manager().enumerate_devices()
 async def query_system_parameters() -> Dict[str, Any]:
    return await (await get_local_device()).query_system_parameters()
 async def spawn(
    program: Union[str, List[Union[str, bytes]]],
    argv: Optional[List[Union[str, bytes]]] = None,
    envp: Optional[Dict[str, str]] = None,
    env: Optional[Dict[str, str]] = None,
    cwd: Optional[str] = None,
    stdio: Optional[str] = None,
    **aux: Any,
 ) -> int:
    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: ProcessTarget) -> None:
    return await (await get_local_device()).resume(target)
 async def kill(target: ProcessTarget) -> None:
    return await (await get_local_device()).kill(target)
 async def attach(target: ProcessTarget, **kwargs: Any) -> "Session":
    return await (await get_local_device()).attach(target, **kwargs)
 async def inject_library_file(target: ProcessTarget, path: str, entrypoint: str, data: str) -> int:
    return await (await get_local_device()).inject_library_file(target, path, entrypoint, data)
 async def inject_library_blob(target: ProcessTarget, blob: bytes, entrypoint: str, data: str) -> int:
    return await (await get_local_device()).inject_library_blob(target, blob, entrypoint, data)
 async def shutdown() -> None:
    await get_device_manager().close()
 def make_auth_callback(callback: Callable[[str], Any]) -> Callable[[Any], str]:
    def authenticate(token: str) -> str:
        session_info = callback(token)
        return json.dumps(session_info)
    return authenticate
 # TEMPORARY DEBUG HACK
 def print_alive_tasks() -> None:
    _frida.print_alive_tasks()
@@ -0,0 +1,106 @@
 ProcessTarget = Union[int, str]
 class ScriptErrorMessage(TypedDict):
    type: Literal["error"]
    description: str
    stack: NotRequired[str]
    fileName: NotRequired[str]
    lineNumber: NotRequired[int]
    columnNumber: NotRequired[int]
 class ScriptPayloadMessage(TypedDict):
    type: Literal["send"]
    payload: NotRequired[Any]
 ScriptMessage = Union[ScriptPayloadMessage, ScriptErrorMessage]
 ScriptMessageCallback = Callable[[ScriptMessage, Optional[bytes]], None]
 ScriptDestroyedCallback = Callable[[], None]
 SessionDetachedCallback = Callable[
    [
        Literal[
            "application-requested", "process-replaced", "process-terminated", "connection-terminated", "device-lost"
        ],
        Optional[_frida.Crash],
    ],
    None,
 ]
 BusDetachedCallback = Callable[[], None]
 BusMessageCallback = Callable[[Mapping[Any, Any], Optional[bytes]], None]
 ServiceCloseCallback = Callable[[], None]
 ServiceMessageCallback = Callable[[Any], None]
 DeviceSpawnAddedCallback = Callable[[_frida.Spawn], None]
 DeviceSpawnRemovedCallback = Callable[[_frida.Spawn], None]
 DeviceChildAddedCallback = Callable[[_frida.Child], None]
 DeviceChildRemovedCallback = Callable[[_frida.Child], None]
 DeviceProcessCrashedCallback = Callable[[_frida.Crash], None]
 DeviceOutputCallback = Callable[[int, int, bytes], None]
 DeviceUninjectedCallback = Callable[[int], None]
 DeviceLostCallback = Callable[[], None]
 DeviceManagerAddedCallback = Callable[[_frida.Device], None]
 DeviceManagerRemovedCallback = Callable[[_frida.Device], None]
 DeviceManagerChangedCallback = Callable[[], None]
 PortalServiceNodeJoinedCallback = Callable[[int, _frida.Application], None]
 PortalServiceNodeLeftCallback = Callable[[int, _frida.Application], None]
 PortalServiceNodeConnectedCallback = Callable[[int, Tuple[str, int]], None]
 PortalServiceNodeDisconnectedCallback = Callable[[int, Tuple[str, int]], None]
 PortalServiceControllerConnectedCallback = Callable[[int, Tuple[str, int]], None]
 PortalServiceControllerDisconnectedCallback = Callable[[int, Tuple[str, int]], None]
 PortalServiceAuthenticatedCallback = Callable[[int, Mapping[Any, Any]], None]
 PortalServiceSubscribeCallback = Callable[[int], None]
 PortalServiceMessageCallback = Callable[[int, Mapping[Any, Any], Optional[bytes]], None]
 class CompilerDiagnosticFile(TypedDict):
    path: str
    line: int
    character: int
 class CompilerDiagnostic(TypedDict):
    category: str
    code: int
    file: NotRequired[CompilerDiagnosticFile]
    text: str
 CompilerStartingCallback = Callable[[], None]
 CompilerFinishedCallback = Callable[[], None]
 CompilerOutputCallback = Callable[[str], None]
 CompilerDiagnosticsCallback = Callable[[List[CompilerDiagnostic]], None]
 CompilerOutputFormat = Literal["unescaped", "hex-bytes", "c-string"]
 CompilerBundleFormat = Literal["esm", "iife"]
 CompilerTypeCheck = Literal["full", "none"]
 CompilerSourceMaps = Literal["included", "omitted"]
 CompilerCompression = Literal["none", "terser"]
 CompilerPlatform = Literal["neutral", "gum", "browser"]
 PackageManagerInstallProgressCallback = Callable[
    [
        Literal[
            "initializing",
            "preparing-dependencies",
            "resolving-package",
            "fetching-resource",
            "package-already-installed",
            "downloading-package",
            "package-installed",
            "resolving-and-installing-all",
            "complete",
        ],
        float,
        Optional[str],
    ],
    None,
 ]
 PackageRole = Literal["runtime", "development", "optional", "peer"]
@@ -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;
 }
@@ -0,0 +1,11 @@
 /* TEMPORARY DEBUG HACK: needs a GLib built with G_ENABLE_DEBUG. */
 extern void g_task_print_alive_tasks (void);
 static PyObject *
 PyFrida_print_alive_tasks (PyObject * module,
                           PyObject * args)
 {
  g_task_print_alive_tasks ();
  Py_RETURN_NONE;
 }
@@ -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
@@ -0,0 +1,409 @@
 from __future__ import annotations
 from typing import Mapping
 from .model import (
    ConstructorCustomizations,
    CustomCode,
    Customizations,
    EnumerationCustomizations,
    EnumerationMemberCustomizations,
    MethodCustomizations,
    ModuleFunction,
    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",
                    ),
                    # TEMPORARY DEBUG HACK
                    ModuleFunction(
                        py_name="print_alive_tasks",
                        c_symbol="PyFrida_print_alive_tasks",
                        asset="module_print_alive_tasks.c",
                    ),
                ),
            ),
        ),
        "Bus": ObjectTypeCustomizations(
            provides_signals=True,
            custom_code=CustomCode(
                members=("facade_bus.py", "facade_bus_aio.py"),
            ),
            methods={
                "post": MethodCustomizations(
                    param_typings=["message: Any", "data: Optional[bytes] = 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: int", "message: Any", "data: Optional[bytes] = None"],
                    custom_logic="json = _to_json(message)",
                ),
                "narrowcast": MethodCustomizations(
                    param_typings=["tag: str", "message: Any", "data: Optional[bytes] = None"],
                    custom_logic="json = _to_json(message)",
                ),
                "broadcast": MethodCustomizations(
                    param_typings=["message: Any", "data: Optional[bytes] = None"],
                    custom_logic="json = _to_json(message)",
                ),
            },
        ),
        "Device": ObjectTypeCustomizations(
            methods={
                "spawn": MethodCustomizations(
                    param_typings=[
                        "program: Union[str, List[Union[str, bytes]]]",
                        "argv: Optional[List[Union[str, bytes]]] = None",
                        "envp: Optional[Mapping[str, str]] = None",
                        "env: Optional[Mapping[str, str]] = None",
                        "cwd: Optional[str] = None",
                        "stdio: Optional[str] = None",
                        "**aux: Any",
                    ],
                    custom_logic="""\
 if not isinstance(program, str):
    args = list(program)
    argv = args if len(args) > 1 else None
    first = args[0]
    program = first.decode() if isinstance(first, bytes) else first
 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: ProcessTarget"], custom_logic=_RESOLVE_PID),
                "kill": MethodCustomizations(param_typings=["target: ProcessTarget"], custom_logic=_RESOLVE_PID),
                "input": MethodCustomizations(
                    param_typings=["target: ProcessTarget", "data: bytes"], custom_logic=_RESOLVE_PID
                ),
                "inject_library_file": MethodCustomizations(
                    param_typings=["target: ProcessTarget", "path: str", "entrypoint: str", "data: str"],
                    custom_logic=_RESOLVE_PID,
                ),
                "inject_library_blob": MethodCustomizations(
                    param_typings=["target: ProcessTarget", "blob: bytes", "entrypoint: str", "data: str"],
                    custom_logic=_RESOLVE_PID,
                ),
                "attach": MethodCustomizations(
                    param_typings=["target: ProcessTarget", "**kwargs: Any"],
                    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: Any", "data: Optional[bytes] = 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: str",
                        "name: Optional[str] = None",
                        "snapshot: Optional[bytes] = None",
                        "runtime: Optional[str] = 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: bytes",
                        "name: Optional[str] = None",
                        "snapshot: Optional[bytes] = None",
                        "runtime: Optional[str] = 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: str", "name: Optional[str] = None", "runtime: Optional[str] = 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: str",
                        "warmup_script: Optional[str] = None",
                        "runtime: Optional[str] = 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: Optional[str] = None",
                    "port: Optional[int] = None",
                    "certificate: Optional[str] = None",
                    "origin: Optional[str] = None",
                    "authentication: Optional[Tuple[str, Any]] = None",
                    "asset_root: Optional[str] = None",
                    "request_handler: Optional[Callable[..., Any]] = 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))  # type: ignore[attr-defined]
            AuthenticationService.__init__(service)
            auth_service = service._impl
        else:
            raise ValueError("invalid authentication scheme")
    else:
        auth_service = _unwrap(authentication)
 kwargs: Dict[str, Any] = {}
 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,
        facade_preludes=("facade_types.py",),
        facade_epilogues=("facade_epilogue.py",),
    )
@@ -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,
    )
@@ -0,0 +1,351 @@
 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 is_select_method(self) -> bool:
        return self.name.startswith("select_") or self.name.startswith("add_")
    @cached_property
    def select_plural_noun(self) -> str:
        return f"{self.name.split('_', maxsplit=1)[1]}s"
    @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)
    facade_preludes: Tuple[str, ...] = ()
    facade_epilogues: Tuple[str, ...] = ()
@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)
@@ -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',
 )
@@ -32,9 +32,9 @@ endif
 subdir('frida')
-test('frida-python', python,
+#test('frida-python', python,
-  args: ['-m', 'unittest', 'discover'],
+#  args: ['-m', 'unittest', 'discover'],
-  workdir: meson.current_source_dir(),
+#  workdir: meson.current_source_dir(),
-  env: {'PYTHONPATH': meson.current_build_dir() / 'src'},
+#  env: {'PYTHONPATH': meson.current_build_dir() / 'src'},
-  timeout: 30,
+#  timeout: 30,
-)
+#)
@@ -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
@@ -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"],
+        packages=["frida"],
-        package_data={"frida": ["py.typed"], "frida._frida": ["py.typed", "__init__.pyi"]},
+        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, Path(FRIDA_EXTENSION).parent)
 class FridaDemandBuiltExt(build_ext):
@@ -119,11 +122,23 @@ 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, SOURCE_ROOT / "build" / "frida")
 def copy_generated_package(build_lib, generated):
    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__":
@@ -1,6 +1,6 @@
 [wrap-git]
 url = https://github.com/frida/frida-core.git
-revision = 16.6.2
+revision = 17.16.4
 depth = 1
 [provide]
@@ -1,4 +0,0 @@
 from .test_core import TestCore
 from .test_rpc import TestRpc
 __all__ = ["TestCore", "TestRpc"]
@@ -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
@@ -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" \
 	"$@"
@@ -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()
@@ -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()
@@ -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()