subprojects: Prepare for release

Reinstate the cancellable decorator, RPCResult, make_rpc_call_request
and make_auth_callback, and expose ScriptExportsAsync from frida.core
via a new facade_epilogues customization.

frida/frida_bindgen/assets/facade_bus.py

@@ -1,20 +1,20 @@
-def _setup(self):
-    self._message_handlers = []
+def _setup(self) -> None:
+    self._message_handlers: List[BusMessageCallback] = []
     self._impl.on("message", self._on_message)
 
-def on(self, signal, callback):
+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, 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, data):
+def _on_message(self, raw_message: str, data: Optional[bytes]) -> None:
     message = json.loads(raw_message)
     for handler in list(self._message_handlers):
         try:

frida/frida_bindgen/assets/facade_bus_aio.py

@@ -1,26 +1,26 @@
-def _setup(self):
-    self._message_handlers = []
+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, callback):
+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, 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, data):
+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, message, data):
+def _deliver_message(self, handler: BusMessageCallback, message: Mapping[Any, Any], data: Optional[bytes]) -> None:
     try:
         handler(message, data)
     except Exception:

frida/frida_bindgen/assets/facade_cancellable.py

@@ -1,4 +1,4 @@
-def __enter__(self):
+def __enter__(self) -> "Cancellable":
     stack = getattr(_current_cancellable, "stack", None)
     if stack is None:
         stack = []
@@ -6,17 +6,44 @@ def __enter__(self):
     stack.append(self)
     return self
 
-def __exit__(self, *exc):
+def __exit__(self, *exc: Any) -> Literal[False]:
     _current_cancellable.stack.pop()
     return False
 
 @classmethod
-def get_current(cls):
+def get_current(cls) -> "Cancellable":
     return _current_cancellable_get()
 
-def raise_if_cancelled(self):
+def raise_if_cancelled(self) -> None:
     if self._impl.is_cancelled():
         raise _frida.OperationCancelledError("operation was cancelled")
 
-def get_pollfd(self):
+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)

frida/frida_bindgen/assets/facade_cancellable_aio.py

@@ -1,18 +1,45 @@
-def __enter__(self):
+def __enter__(self) -> "Cancellable":
     self._token = _current_cancellable.set(self)
     return self
 
-def __exit__(self, *exc):
+def __exit__(self, *exc: Any) -> Literal[False]:
     _current_cancellable.reset(self._token)
     return False
 
 @classmethod
-def get_current(cls):
+def get_current(cls) -> "Cancellable":
     return _current_cancellable.get()
 
-def raise_if_cancelled(self):
+def raise_if_cancelled(self) -> None:
     if self._impl.is_cancelled():
         raise _frida.OperationCancelledError("operation was cancelled")
 
-def get_pollfd(self):
+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)

frida/frida_bindgen/assets/facade_cancellable_helpers.py

@@ -1,23 +1,40 @@
 class CancellablePollFD:
-    def __init__(self, cancellable):
+    def __init__(self, cancellable: _frida.Cancellable) -> None:
         self.handle = cancellable.get_fd()
-        self._cancellable = cancellable
+        self._cancellable: Optional[_frida.Cancellable] = cancellable
 
-    def __del__(self):
+    def __del__(self) -> None:
         self.release()
 
-    def release(self):
+    def release(self) -> None:
         if self._cancellable is not None:
             if self.handle != -1:
                 self._cancellable.release_fd()
                 self.handle = -1
             self._cancellable = None
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return repr(self.handle)
 
-    def __enter__(self):
+    def __enter__(self) -> int:
         return self.handle
 
-    def __exit__(self, *exc):
+    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

frida/frida_bindgen/assets/facade_cancellable_helpers_aio.py

@@ -1,23 +1,40 @@
 class CancellablePollFD:
-    def __init__(self, cancellable):
+    def __init__(self, cancellable: _frida.Cancellable) -> None:
         self.handle = cancellable.get_fd()
-        self._cancellable = cancellable
+        self._cancellable: Optional[_frida.Cancellable] = cancellable
 
-    def __del__(self):
+    def __del__(self) -> None:
         self.release()
 
-    def release(self):
+    def release(self) -> None:
         if self._cancellable is not None:
             if self.handle != -1:
                 self._cancellable.release_fd()
                 self.handle = -1
             self._cancellable = None
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return repr(self.handle)
 
-    def __enter__(self):
+    def __enter__(self) -> int:
         return self.handle
 
-    def __exit__(self, *exc):
+    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

frida/frida_bindgen/assets/facade_device_manager_members.py

@@ -1,4 +1,4 @@
-def get_device_matching(self, predicate, timeout=0):
+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():
@@ -7,3 +7,19 @@ def get_device_matching(self, predicate, timeout=0):
         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)

frida/frida_bindgen/assets/facade_device_manager_members_aio.py

@@ -1,4 +1,4 @@
-async def get_device_matching(self, predicate, timeout=0):
+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():
@@ -7,3 +7,19 @@ async def get_device_matching(self, predicate, timeout=0):
         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)

frida/frida_bindgen/assets/facade_device_members.py

@@ -1,11 +1,11 @@
 @property
-def type(self):
+def type(self) -> str:
     return self.dtype
 
-def get_bus(self):
+def get_bus(self) -> "Bus":
     return self.bus
 
-def get_process(self, process_name):
+def get_process(self, process_name: str) -> "Process":
     process_name_lc = process_name.lower()
     matching = [
         process
@@ -19,7 +19,7 @@ def get_process(self, process_name):
         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):
+def _pid_of(self, target: ProcessTarget) -> int:
     if isinstance(target, str):
         return self.get_process(target).pid
     return target

frida/frida_bindgen/assets/facade_device_members_aio.py

@@ -1,11 +1,11 @@
 @property
-def type(self):
+def type(self) -> str:
     return self.dtype
 
-def get_bus(self):
+def get_bus(self) -> "Bus":
     return self.bus
 
-async def get_process(self, process_name):
+async def get_process(self, process_name: str) -> "Process":
     process_name_lc = process_name.lower()
     matching = [
         process
@@ -19,7 +19,7 @@ async def get_process(self, process_name):
         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):
+async def _pid_of(self, target: ProcessTarget) -> int:
     if isinstance(target, str):
         return (await self.get_process(target)).pid
     return target

frida/frida_bindgen/assets/facade_epilogue.py

@@ -0,0 +1 @@
+ScriptExportsAsync = aio._ScriptExports

frida/frida_bindgen/assets/facade_interface.py

@@ -1,6 +1,6 @@
 class _Implementation:
-    def _frida_dispatch(self, name, args, completion):
-        def run():
+    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

frida/frida_bindgen/assets/facade_interface_aio.py

@@ -1,8 +1,10 @@
 class _Implementation:
-    def _frida_dispatch(self, name, args, completion):
+    _loop: Any
+
+    def _frida_dispatch(self, name: str, args: List[Any], completion: Any) -> None:
         loop = self._loop
 
-        async def run():
+        async def run() -> None:
             try:
                 result = getattr(self, name)(*[_wrap(a) for a in args])
                 if asyncio.iscoroutine(result):

frida/frida_bindgen/assets/facade_iostream.py

@@ -1,10 +1,10 @@
-def close(self):
+def close(self) -> None:
     _invoke(self._impl.close_async, (0,))
 
-def read(self, count):
+def read(self, count: int) -> bytes:
     return _invoke(self._impl.input_stream.read_bytes_async, (count, 0))
 
-def read_all(self, count):
+def read_all(self, count: int) -> bytes:
     chunks = []
     remaining = count
     while remaining > 0:
@@ -15,10 +15,10 @@ def read_all(self, count):
         remaining -= len(chunk)
     return b"".join(chunks)
 
-def write(self, data):
+def write(self, data: bytes) -> int:
     return _invoke(self._impl.output_stream.write_bytes_async, (data, 0))
 
-def write_all(self, data):
+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))

frida/frida_bindgen/assets/facade_iostream_aio.py

@@ -1,10 +1,10 @@
-async def close(self):
+async def close(self) -> None:
     await _invoke(self._impl.close_async, (0,))
 
-async def read(self, count):
+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):
+async def read_all(self, count: int) -> bytes:
     chunks = []
     remaining = count
     while remaining > 0:
@@ -15,10 +15,10 @@ async def read_all(self, count):
         remaining -= len(chunk)
     return b"".join(chunks)
 
-async def write(self, data):
+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):
+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))

frida/frida_bindgen/assets/facade_portal.py

@@ -1,10 +1,10 @@
-def _setup(self):
-    self._authenticated_handlers = []
-    self._message_handlers = []
+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, callback):
+def on(self, signal: str, callback: Callable[..., Any]) -> None:
     if signal == "authenticated":
         self._authenticated_handlers.append(callback)
     elif signal == "message":
@@ -12,7 +12,7 @@ def on(self, signal, callback):
     else:
         self._impl.on(signal, _make_signal_handler(callback))
 
-def off(self, signal, callback):
+def off(self, signal: str, callback: Callable[..., Any]) -> None:
     if signal == "authenticated":
         self._authenticated_handlers.remove(callback)
     elif signal == "message":
@@ -20,7 +20,7 @@ def off(self, signal, callback):
     else:
         self._impl.off(signal, callback)
 
-def _on_authenticated(self, connection_id, raw_session_info):
+def _on_authenticated(self, connection_id: int, raw_session_info: str) -> None:
     session_info = json.loads(raw_session_info)
     for handler in list(self._authenticated_handlers):
         try:
@@ -28,7 +28,7 @@ def _on_authenticated(self, connection_id, raw_session_info):
         except Exception:
             traceback.print_exc()
 
-def _on_message(self, connection_id, raw_message, data):
+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:

frida/frida_bindgen/assets/facade_portal_aio.py

@@ -1,11 +1,11 @@
-def _setup(self):
-    self._authenticated_handlers = []
-    self._message_handlers = []
+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, callback):
+def on(self, signal: str, callback: Callable[..., Any]) -> None:
     if signal == "authenticated":
         self._authenticated_handlers.append(callback)
     elif signal == "message":
@@ -13,7 +13,7 @@ def on(self, signal, callback):
     else:
         self._impl.on(signal, _make_signal_handler(callback))
 
-def off(self, signal, callback):
+def off(self, signal: str, callback: Callable[..., Any]) -> None:
     if signal == "authenticated":
         self._authenticated_handlers.remove(callback)
     elif signal == "message":
@@ -21,17 +21,17 @@ def off(self, signal, callback):
     else:
         self._impl.off(signal, callback)
 
-def _on_authenticated(self, connection_id, raw_session_info):
+def _on_authenticated(self, connection_id: int, raw_session_info: str) -> None:
     session_info = json.loads(raw_session_info)
     for handler in list(self._authenticated_handlers):
         _dispatch(self._loop, self._deliver, handler, connection_id, session_info)
 
-def _on_message(self, connection_id, raw_message, data):
+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, *args):
+def _deliver(self, handler: Callable[..., Any], *args: Any) -> None:
     try:
         handler(*args)
     except Exception:

frida/frida_bindgen/assets/facade_rpc_helpers.py

@@ -1,4 +1,4 @@
-def _to_camel_case(name):
+def _to_camel_case(name: str) -> str:
     result = ""
     capitalize = False
     for ch in name:
@@ -13,19 +13,19 @@ def _to_camel_case(name):
 
 
 class RPCException(Exception):
-    def __str__(self):
+    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):
+    def __init__(self, script: "Script") -> None:
         self._script = script
 
-    def __getattr__(self, name):
+    def __getattr__(self, name: str) -> Callable[..., Any]:
         script = self._script
         js_name = _to_camel_case(name)
 
-        def method(*args):
+        def method(*args: Any) -> Any:
             if args and isinstance(args[-1], bytes):
                 params, data = list(args[:-1]), args[-1]
             else:
@@ -34,5 +34,26 @@ class _ScriptExports:
 
         return method
 
-    def __dir__(self):
+    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)

frida/frida_bindgen/assets/facade_rpc_helpers_aio.py

@@ -1,4 +1,4 @@
-def _to_camel_case(name):
+def _to_camel_case(name: str) -> str:
     result = ""
     capitalize = False
     for ch in name:
@@ -13,19 +13,19 @@ def _to_camel_case(name):
 
 
 class RPCException(Exception):
-    def __str__(self):
+    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):
+    def __init__(self, script: "Script") -> None:
         self._script = script
 
-    def __getattr__(self, name):
+    def __getattr__(self, name: str) -> Callable[..., Any]:
         script = self._script
         js_name = _to_camel_case(name)
 
-        def method(*args):
+        def method(*args: Any) -> Any:
             if args and isinstance(args[-1], bytes):
                 params, data = list(args[:-1]), args[-1]
             else:
@@ -33,3 +33,23 @@ class _ScriptExports:
             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)

frida/frida_bindgen/assets/facade_rpc_members.py

@@ -1,55 +1,55 @@
-def _setup(self):
+def _setup(self) -> None:
     self.exports_sync = _ScriptExports(self)
-    self._message_handlers = []
-    self._log_handler = self.default_log_handler
-    self._pending = {}
+    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):
+def exports(self) -> "_ScriptExports":
     return self.exports_sync
 
-def on(self, signal, callback):
+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, 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):
+def get_log_handler(self) -> Callable[[str, str], None]:
     return self._log_handler
 
-def set_log_handler(self, handler):
+def set_log_handler(self, handler: Callable[[str, str], None]) -> None:
     self._log_handler = handler
 
-def default_log_handler(self, level, text):
+def default_log_handler(self, level: str, text: str) -> None:
     if level == "info":
         print(text, file=sys.stdout)
     else:
         print(text, file=sys.stderr)
 
-def list_exports_sync(self):
+def list_exports_sync(self) -> List[str]:
     return self._rpc_request(["list"])
 
-def list_exports(self):
+def list_exports(self) -> List[str]:
     return self.list_exports_sync()
 
-def _list_exports(self):
+def _list_exports(self) -> List[str]:
     return self.list_exports_sync()
 
-def _rpc_request(self, args, data=None):
-    outcome = {}
+def _rpc_request(self, args: Any, data: Optional[bytes] = None) -> Any:
+    outcome: Dict[str, Any] = {}
     cond = self._cond
 
-    def complete(value, error):
+    def complete(value: Any, error: Optional[Exception]) -> None:
         with cond:
             outcome["value"] = value
             outcome["error"] = error
@@ -73,7 +73,7 @@ def _rpc_request(self, args, data=None):
         raise outcome["error"]
     return outcome["value"]
 
-def _on_message(self, raw_message, data):
+def _on_message(self, raw_message: str, data: Optional[bytes]) -> None:
     message = json.loads(raw_message)
     mtype = message["type"]
     payload = message.get("payload")
@@ -88,7 +88,7 @@ def _on_message(self, raw_message, data):
             except Exception:
                 traceback.print_exc()
 
-def _on_destroyed(self):
+def _on_destroyed(self) -> None:
     while True:
         with self._cond:
             pending_ids = list(self._pending.keys())
@@ -97,7 +97,7 @@ def _on_destroyed(self):
             break
         complete(None, _frida.InvalidOperationError("script has been destroyed"))
 
-def _on_rpc_message(self, request_id, operation, params, data):
+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)

frida/frida_bindgen/assets/facade_rpc_members_aio.py

@@ -1,45 +1,45 @@
-def _setup(self):
+def _setup(self) -> None:
     self.exports = _ScriptExports(self)
-    self._message_handlers = []
-    self._log_handler = self.default_log_handler
-    self._pending = {}
+    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, callback):
+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, 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):
+def get_log_handler(self) -> Callable[[str, str], None]:
     return self._log_handler
 
-def set_log_handler(self, handler):
+def set_log_handler(self, handler: Callable[[str, str], None]) -> None:
     self._log_handler = handler
 
-def default_log_handler(self, level, text):
+def default_log_handler(self, level: str, text: str) -> None:
     if level == "info":
         print(text, file=sys.stdout)
     else:
         print(text, file=sys.stderr)
 
-async def list_exports(self):
+async def list_exports(self) -> List[str]:
     return await self._rpc_request(["list"])
 
-def _rpc_request(self, args, data=None):
+def _rpc_request(self, args: Any, data: Optional[bytes] = None) -> "asyncio.Future[Any]":
     loop = asyncio.get_running_loop()
-    future = loop.create_future()
+    future: "asyncio.Future[Any]" = loop.create_future()
 
-    def complete(value, error):
+    def complete(value: Any, error: Optional[Exception]) -> None:
         if future.done():
             return
         if error is not None:
@@ -58,7 +58,7 @@ def _rpc_request(self, args, data=None):
 
     return future
 
-def _on_message(self, raw_message, data):
+def _on_message(self, raw_message: str, data: Optional[bytes]) -> None:
     message = json.loads(raw_message)
     mtype = message["type"]
     payload = message.get("payload")
@@ -70,13 +70,13 @@ def _on_message(self, raw_message, data):
         for handler in list(self._message_handlers):
             _dispatch(self._loop, self._deliver_message, handler, message, data)
 
-def _deliver_message(self, 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):
+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
@@ -84,7 +84,7 @@ def _on_destroyed(self):
             break
         complete(None, _frida.InvalidOperationError("script has been destroyed"))
 
-def _on_rpc_message(self, request_id, operation, params, data):
+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)

frida/frida_bindgen/assets/facade_toplevel.py

@@ -1,63 +1,79 @@
 __version__ = _frida.__version__
 
 
-def get_device_manager():
+def get_device_manager() -> "DeviceManager":
     return _wrap(_frida.get_device_manager())
 
 
-def get_device(id, timeout=0):
+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, timeout=0):
+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():
+def get_local_device() -> "Device":
     return get_device_manager().get_device_by_type("local", 0)
 
 
-def get_remote_device():
+def get_remote_device() -> "Device":
     return get_device_manager().get_device_by_type("remote", 0)
 
 
-def get_usb_device(timeout=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():
+def enumerate_devices() -> List["Device"]:
     return get_device_manager().enumerate_devices()
 
 
-def query_system_parameters():
+def query_system_parameters() -> Dict[str, Any]:
     return get_local_device().query_system_parameters()
 
 
-def spawn(program, argv=None, envp=None, env=None, cwd=None, stdio=None, **aux):
+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):
+def resume(target: ProcessTarget) -> None:
     return get_local_device().resume(target)
 
 
-def kill(target):
+def kill(target: ProcessTarget) -> None:
     return get_local_device().kill(target)
 
 
-def attach(target, **kwargs):
+def attach(target: ProcessTarget, **kwargs: Any) -> "Session":
     return get_local_device().attach(target, **kwargs)
 
 
-def inject_library_file(target, path, entrypoint, data):
+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, blob, 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():
+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

frida/frida_bindgen/assets/facade_toplevel_aio.py

@@ -1,64 +1,80 @@
 __version__ = _frida.__version__
 
 
-def get_device_manager():
+def get_device_manager() -> "DeviceManager":
     return _wrap(_frida.get_device_manager())
 
 
-async def get_device(id, timeout=0):
+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, timeout=0):
+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():
+async def get_local_device() -> "Device":
     return await get_device_manager().get_device_by_type("local", 0)
 
 
-async def get_remote_device():
+async def get_remote_device() -> "Device":
     return await get_device_manager().get_device_by_type("remote", 0)
 
 
-async def get_usb_device(timeout=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():
+async def enumerate_devices() -> List["Device"]:
     return await get_device_manager().enumerate_devices()
 
 
-async def query_system_parameters():
+async def query_system_parameters() -> Dict[str, Any]:
     return await (await get_local_device()).query_system_parameters()
 
 
-async def spawn(program, argv=None, envp=None, env=None, cwd=None, stdio=None, **aux):
+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):
+async def resume(target: ProcessTarget) -> None:
     return await (await get_local_device()).resume(target)
 
 
-async def kill(target):
+async def kill(target: ProcessTarget) -> None:
     return await (await get_local_device()).kill(target)
 
 
-async def attach(target, **kwargs):
+async def attach(target: ProcessTarget, **kwargs: Any) -> "Session":
     return await (await get_local_device()).attach(target, **kwargs)
 
 
-async def inject_library_file(target, path, entrypoint, data):
+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, blob, 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():
+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

frida/frida_bindgen/assets/facade_types.py

@@ -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"]

frida/frida_bindgen/codegen.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
+import ast
+import re
 import textwrap
 from pathlib import Path
-from typing import Dict, List, Optional, Tuple, Union
+from typing import Dict, List, Optional, Set, Tuple, Union
 
 from frida_bindgen_core import Procedure, Type
+from frida_bindgen_core.naming import to_pascal_case
 
 from .model import (
     Enumeration,
@@ -31,6 +34,19 @@ def read_asset(name: str) -> str:
     return (ASSETS_DIR / name).read_text(encoding="utf-8")
 
 
+FACADE_TYPING_IMPORTS = (
+    "from typing import (Any, Callable, Dict, List, Literal, Mapping, NotRequired, Optional, Tuple, TypedDict, "
+    "Sequence, Union, ParamSpec, TypeVar, cast, overload)"
+)
+
+
+def generate_facade_preludes(model: Model) -> List[str]:
+    lines = []
+    for asset in model.customizations.facade_preludes:
+        lines += ["", read_asset(asset).strip(), ""]
+    return lines
+
+
 FACADE_RUNTIME = """
 _to_json = json.dumps
 
@@ -179,7 +195,7 @@ def _make_options(cls, values, selectors):
     return options
 
 
-_current_cancellable = contextvars.ContextVar("frida_current_cancellable", default=None)
+_current_cancellable: contextvars.ContextVar = contextvars.ContextVar("frida_current_cancellable", default=None)
 
 
 def _dispatch(loop, callback, *args):
@@ -228,16 +244,20 @@ def generate_all(model: Model) -> Dict[str, str]:
 def generate_py(model: Model) -> str:
     lines = [
         "from . import _frida",
+        "import dataclasses",
         "import fnmatch",
+        "import functools",
         "import inspect",
         "import time",
         "import json",
         "import sys",
         "import threading",
         "import traceback",
+        FACADE_TYPING_IMPORTS,
         "",
         "",
         *generate_py_exception_reexports(model),
+        *generate_facade_preludes(model),
         "",
         FACADE_RUNTIME.strip(),
         "",
@@ -262,6 +282,9 @@ def generate_py(model: Model) -> str:
     lines.append("")
     lines.append("")
     lines.append("from . import aio")
+    for asset in model.customizations.facade_epilogues:
+        lines.append("")
+        lines.append(read_asset(asset).strip())
     lines.append("")
     lines.append("core = sys.modules[__name__]")
     lines.append('sys.modules[__name__ + ".core"] = core')
@@ -294,14 +317,18 @@ def generate_aio(model: Model) -> str:
         "import asyncio",
         "import contextvars",
         "import time",
+        "import dataclasses",
         "import fnmatch",
+        "import functools",
         "import inspect",
         "import json",
         "import sys",
         "import traceback",
+        FACADE_TYPING_IMPORTS,
         "",
         "",
         *generate_py_exception_reexports(model),
+        *generate_facade_preludes(model),
         "",
         AIO_RUNTIME.strip(),
         "",
@@ -350,6 +377,8 @@ def generate_aio_class(otype: ObjectType, model: Model) -> str:
     if not members:
         members.append("    pass")
 
+    members = apply_signal_overloads(members, otype, model)
+
     return f"class {otype.py_name}:\n" + "\n\n".join(members)
 
 
@@ -373,7 +402,7 @@ def generate_aio_method(method: Method, model: Model) -> Optional[str]:
 
 def generate_facade_bool_property(method: Method) -> str:
     return f"""    @property
-    def {method.name}(self):
+    def {method.name}(self) -> bool:
         return self._impl.{method.name}()"""
 
 
@@ -387,7 +416,9 @@ def generate_aio_async_method(method: Method, model: Model) -> Optional[str]:
     if method.return_value is not None:
         call = wrap_result(call, method.return_value.type, model)
 
-    return f"""    async def {method.name}({signature}):
+    ret = "None" if method.return_value is None else pyi_type(method.return_value.type, model)
+
+    return f"""    async def {method.name}({signature}) -> {ret}:
         return {call}"""
 
 
@@ -408,7 +439,9 @@ def generate_custom_facade_method(
     body = textwrap.indent(logic.strip(), " " * 8)
     keyword = "async def" if awaitable else "def"
 
-    return f"""    {keyword} {method.name}({signature}):
+    ret = "None" if method.return_value is None else pyi_type(method.return_value.type, model)
+
+    return f"""    {keyword} {method.name}({signature}) -> {ret}:
 {body}
         return {call}"""
 
@@ -441,6 +474,8 @@ def generate_py_class(otype: ObjectType, model: Model) -> str:
     if not members:
         members.append("    pass")
 
+    members = apply_signal_overloads(members, otype, model)
+
     return f"class {otype.py_name}:\n" + "\n\n".join(members)
 
 
@@ -476,13 +511,54 @@ def generate_facade_init(otype: ObjectType) -> str:
 
 
 def generate_facade_signals() -> str:
-    return """    def on(self, signal, callback):
+    return """    def on(self, signal: str, callback: Callable[..., Any]) -> None:
         self._impl.on(signal, _make_signal_handler(callback))
 
-    def off(self, signal, callback):
+    def off(self, signal: str, callback: Callable[..., Any]) -> None:
         self._impl.off(signal, callback)"""
 
 
+def facade_prelude_names(model: Model) -> Set[str]:
+    names = set()
+    for asset in model.customizations.facade_preludes:
+        for node in ast.parse(read_asset(asset)).body:
+            if isinstance(node, ast.ClassDef):
+                names.add(node.name)
+            elif isinstance(node, ast.Assign):
+                names.update(t.id for t in node.targets if isinstance(t, ast.Name))
+    return names
+
+
+def signal_callback_alias(otype: ObjectType, signal) -> str:
+    return f"{otype.py_name}{to_pascal_case(signal.name.replace('-', '_'))}Callback"
+
+
+def signal_overload_block(otype: ObjectType, model: Model, name: str) -> Optional[str]:
+    aliases = facade_prelude_names(model)
+    lines = []
+    for signal in otype.signals:
+        alias = signal_callback_alias(otype, signal)
+        if alias in aliases:
+            lines.append("    @overload")
+            lines.append(f'    def {name}(self, signal: Literal["{signal.name}"], callback: {alias}) -> None: ...')
+    return "\n".join(lines) if len(lines) > 2 else None
+
+
+def apply_signal_overloads(members: List[str], otype: ObjectType, model: Model) -> List[str]:
+    result = []
+    for member in members:
+        for name in ("on", "off"):
+            block = signal_overload_block(otype, model, name)
+            if block is None:
+                continue
+            pattern = re.compile(rf"^    (?:async )?def {name}\(self, signal", re.MULTILINE)
+            match = pattern.search(member)
+            if match is not None:
+                member = member[: match.start()] + block + "\n" + member[match.start() :]
+        result.append(member)
+    return result
+
+
 def facade_repr_property_names(otype: ObjectType) -> List[str]:
     names = []
     for method in otype.methods:
@@ -516,8 +592,9 @@ def generate_py_property(method: Method, model: Model) -> Optional[str]:
         return None
 
     access = wrap_result(f"self._impl.{name}", method.return_value.type, model)
+    ret = pyi_type(method.return_value.type, model)
     prop = f"""    @property
-    def {name}(self):
+    def {name}(self) -> {ret}:
         return {access}"""
 
     set_method = next((m for m in method.object_type.methods if m.name == f"set_{name}"), None)
@@ -525,7 +602,7 @@ def generate_py_property(method: Method, model: Model) -> Optional[str]:
         prop += f"""
 
     @{name}.setter
-    def {name}(self, value):
+    def {name}(self, value: {ret}) -> None:
         self._impl.{name} = _unwrap(value)"""
 
     return prop
@@ -552,17 +629,26 @@ def generate_py_sync_method(method: Method, model: Model) -> Optional[str]:
 
     signature = ["self"]
     for param in method.input_parameters:
-        signature.append(f"{param.name}=None" if param.nullable else param.name)
+        signature.append(facade_param(param, model))
     names = ", ".join(param.name for param in method.input_parameters)
 
     call = f"self._impl.{method.name}({names})"
     if method.return_value is not None:
         call = wrap_result(call, method.return_value.type, model)
 
-    return f"""    def {method.name}({", ".join(signature)}):
+    ret = "None" if method.return_value is None else pyi_type(method.return_value.type, model)
+
+    return f"""    def {method.name}({", ".join(signature)}) -> {ret}:
         return {call}"""
 
 
+def facade_param(param, model: Model) -> str:
+    annotation = pyi_type(param.type, model)
+    if param.nullable:
+        return f"{param.name}: Optional[{annotation}] = None"
+    return f"{param.name}: {annotation}"
+
+
 def generate_py_async_method(method: Method, model: Model) -> Optional[str]:
     if method.is_property_accessor:
         return None
@@ -576,7 +662,9 @@ def generate_py_async_method(method: Method, model: Model) -> Optional[str]:
     if method.return_value is not None:
         call = wrap_result(call, method.return_value.type, model)
 
-    return f"""    def {method.name}({signature}):
+    ret = "None" if method.return_value is None else pyi_type(method.return_value.type, model)
+
+    return f"""    def {method.name}({signature}) -> {ret}:
         return {call}"""
 
 
@@ -593,10 +681,10 @@ def build_facade_async_parts(method: Method, model: Model) -> Optional[Tuple[str
             signature.append("**kwargs")
             args.append(f"_make_options(_frida.{options.py_name}, kwargs, {build_option_selectors(options)})")
         elif resolve_input_object_type(param.type, model) is not None:
-            signature.append(f"{param.name}=None")
+            signature.append(f"{param.name}: Optional[{pyi_type(param.type, model)}] = None")
             args.append(f"_unwrap({param.name})")
         else:
-            signature.append(param.name)
+            signature.append(facade_param(param, model))
             args.append(param.name)
 
     if method.return_value is not None:
@@ -645,6 +733,11 @@ def generate_extension_pyi(model: Model) -> str:
         lines += generate_pyi_class(otype, model)
 
     lines.append("")
+    lines.append("")
+    for fn in module_functions(model):
+        lines.append(f"def {fn.py_name}() -> Any: ...")
+    lines.append("def _complete_request(request: Any, result: Any, error: Any) -> None: ...")
+    lines.append("")
 
     return "\n".join(lines)
 
@@ -670,8 +763,13 @@ def generate_pyi_class(otype: ObjectType, model: Model) -> List[str]:
         name = property_name_from_accessor(method)
         if name is None or property_getter_marshal(method) is None:
             continue
+        typing = pyi_type(method.return_value.type, model)
         body.append("    @property")
-        body.append(f"    def {name}(self) -> {pyi_type(method.return_value.type, model)}: ...")
+        body.append(f"    def {name}(self) -> {typing}: ...")
+        set_method = next((m for m in otype.methods if m.name == f"set_{name}"), None)
+        if set_method is not None and property_setter_supported(set_method):
+            body.append(f"    @{name}.setter")
+            body.append(f"    def {name}(self, value: {typing}) -> None: ...")
 
     for method in otype.methods:
         stub = pyi_method(method, model)

frida/frida_bindgen/customization.py

@@ -44,7 +44,7 @@ def load_customizations() -> Customizations:
             ),
             methods={
                 "post": MethodCustomizations(
-                    param_typings=["message", "data=None"],
+                    param_typings=["message: Any", "data: Optional[bytes] = None"],
                     custom_logic="json = _to_json(message)",
                 ),
             },
@@ -56,15 +56,15 @@ def load_customizations() -> Customizations:
             ),
             methods={
                 "post": MethodCustomizations(
-                    param_typings=["connection_id", "message", "data=None"],
+                    param_typings=["connection_id: int", "message: Any", "data: Optional[bytes] = None"],
                     custom_logic="json = _to_json(message)",
                 ),
                 "narrowcast": MethodCustomizations(
-                    param_typings=["tag", "message", "data=None"],
+                    param_typings=["tag: str", "message: Any", "data: Optional[bytes] = None"],
                     custom_logic="json = _to_json(message)",
                 ),
                 "broadcast": MethodCustomizations(
-                    param_typings=["message", "data=None"],
+                    param_typings=["message: Any", "data: Optional[bytes] = None"],
                     custom_logic="json = _to_json(message)",
                 ),
             },
@@ -73,22 +73,20 @@ def load_customizations() -> Customizations:
             methods={
                 "spawn": MethodCustomizations(
                     param_typings=[
-                        "program",
-                        "argv=None",
-                        "envp=None",
-                        "env=None",
-                        "cwd=None",
-                        "stdio=None",
-                        "**aux",
+                        "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):
-    argv = program
-    program = argv[0]
-    if len(argv) == 1:
-        argv = None
-if isinstance(program, bytes):
-    program = program.decode()
+    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:
@@ -106,19 +104,21 @@ if aux:
 """,
                 ),
                 "is_lost": MethodCustomizations(as_property=True),
-                "resume": MethodCustomizations(param_typings=["target"], custom_logic=_RESOLVE_PID),
-                "kill": MethodCustomizations(param_typings=["target"], custom_logic=_RESOLVE_PID),
-                "input": MethodCustomizations(param_typings=["target", "data"], custom_logic=_RESOLVE_PID),
+                "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", "path", "entrypoint", "data"],
+                    param_typings=["target: ProcessTarget", "path: str", "entrypoint: str", "data: str"],
                     custom_logic=_RESOLVE_PID,
                 ),
                 "inject_library_blob": MethodCustomizations(
-                    param_typings=["target", "blob", "entrypoint", "data"],
+                    param_typings=["target: ProcessTarget", "blob: bytes", "entrypoint: str", "data: str"],
                     custom_logic=_RESOLVE_PID,
                 ),
                 "attach": MethodCustomizations(
-                    param_typings=["target", "**kwargs"],
+                    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"
@@ -138,7 +138,7 @@ if aux:
             ),
             methods={
                 "post": MethodCustomizations(
-                    param_typings=["message", "data=None"],
+                    param_typings=["message: Any", "data: Optional[bytes] = None"],
                     custom_logic="json = _to_json(message)",
                 ),
                 "is_destroyed": MethodCustomizations(as_property=True),
@@ -148,7 +148,12 @@ if aux:
             methods={
                 "is_detached": MethodCustomizations(as_property=True),
                 "create_script": MethodCustomizations(
-                    param_typings=["source", "name=None", "snapshot=None", "runtime=None"],
+                    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:
@@ -159,7 +164,12 @@ if runtime is not None:
     options.runtime = runtime""",
                 ),
                 "create_script_from_bytes": MethodCustomizations(
-                    param_typings=["data", "name=None", "snapshot=None", "runtime=None"],
+                    param_typings=[
+                        "data: bytes",
+                        "name: Optional[str] = None",
+                        "snapshot: Optional[bytes] = None",
+                        "runtime: Optional[str] = None",
+                    ],
                     custom_logic="""\
 bytes = data
 options = _frida.ScriptOptions()
@@ -171,7 +181,7 @@ if runtime is not None:
     options.runtime = runtime""",
                 ),
                 "compile_script": MethodCustomizations(
-                    param_typings=["source", "name=None", "runtime=None"],
+                    param_typings=["source: str", "name: Optional[str] = None", "runtime: Optional[str] = None"],
                     custom_logic="""\
 options = _frida.ScriptOptions()
 if name is not None:
@@ -180,7 +190,11 @@ if runtime is not None:
     options.runtime = runtime""",
                 ),
                 "snapshot_script": MethodCustomizations(
-                    param_typings=["embed_script", "warmup_script=None", "runtime=None"],
+                    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:
@@ -199,13 +213,13 @@ if runtime is not None:
             custom_constructor="codegen_endpoint_parameters.c",
             constructor=ConstructorCustomizations(
                 param_typings=[
-                    "address=None",
-                    "port=None",
-                    "certificate=None",
-                    "origin=None",
-                    "authentication=None",
-                    "asset_root=None",
-                    "request_handler=None",
+                    "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
@@ -218,7 +232,7 @@ if authentication is not None:
             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))
+            service.authenticate = lambda token: json.dumps(data(token))  # type: ignore[attr-defined]
             AuthenticationService.__init__(service)
             auth_service = service._impl
         else:
@@ -226,7 +240,7 @@ if authentication is not None:
     else:
         auth_service = _unwrap(authentication)
 
-kwargs = {}
+kwargs: Dict[str, Any] = {}
 if address is not None:
     kwargs["address"] = address
 if port is not None:
@@ -382,4 +396,8 @@ if request_handler is not None:
         ),
     }
 
-    return Customizations(type_customizations=type_customizations)
+    return Customizations(
+        type_customizations=type_customizations,
+        facade_preludes=("facade_types.py",),
+        facade_epilogues=("facade_epilogue.py",),
+    )

frida/frida_bindgen/model.py

@@ -200,6 +200,8 @@ class EnumerationMember(core.EnumerationMember):
 @dataclass
 class Customizations:
     type_customizations: Mapping[str, TypeCustomizations] = field(default_factory=OrderedDict)
+    facade_preludes: Tuple[str, ...] = ()
+    facade_epilogues: Tuple[str, ...] = ()
 
 
 @dataclass

subprojects/frida-core.wrap

@@ -1,6 +1,6 @@
 [wrap-git]
 url = https://github.com/frida/frida-core.git
-revision = 17.16.3
+revision = 17.16.4
 depth = 1
 
 [provide]