mirror of
https://github.com/frida/frida-python.git
synced 2026-11-11 10:01:27 -10:00
Compare commits
No files matched your search
@@ -1,4 +1,4 @@
|
||||
PYTHON ?= $(shell which python3 >/dev/null && echo python3 || echo python)
|
||||
PYTHON ?= $(shell command -v python3 >/dev/null && echo python3 || echo python)
|
||||
|
||||
all $(MAKECMDGOALS):
|
||||
@$(PYTHON) \
|
||||
@@ -8,7 +8,7 @@ all $(MAKECMDGOALS):
|
||||
$(MAKECMDGOALS)
|
||||
|
||||
git-submodules:
|
||||
@if [ ! -f releng/meson/meson.py ] || [ ! -f frida-bindgen/frida_bindgen_core/model.py ]; then \
|
||||
@if [ ! -f releng/meson/meson.py ] || git submodule status 2>/dev/null | grep -q '^-'; then \
|
||||
git submodule update --init --recursive --depth 1; \
|
||||
fi
|
||||
-include git-submodules
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
#!/bin/sh
|
||||
|
||||
[ -z "$PYTHON" ] && PYTHON=$(which python3 >/dev/null && echo python3 || echo python)
|
||||
[ -z "$PYTHON" ] && PYTHON=$(command -v python3 >/dev/null && echo python3 || echo python)
|
||||
|
||||
cd $(dirname $0)
|
||||
|
||||
srcroot=$(pwd)
|
||||
|
||||
if [ ! -f releng/meson/meson.py ]; then
|
||||
if [ ! -f releng/meson/meson.py ] || git submodule status 2>/dev/null | grep -q '^-'; then
|
||||
git submodule update --init --recursive --depth 1 || exit $?
|
||||
fi
|
||||
|
||||
|
||||
Generated
+4
-4
@@ -9,7 +9,7 @@
|
||||
"version": "1.0.0",
|
||||
"dependencies": {
|
||||
"core-js": "^3.6.5",
|
||||
"dbus-next": "frida/node-dbus-next",
|
||||
"dbus-next": "frida-forks/node-dbus-next",
|
||||
"duplexify": "^4.1.1",
|
||||
"vue": "^2.6.11",
|
||||
"vue-class-component": "^7.2.3",
|
||||
@@ -6509,7 +6509,7 @@
|
||||
},
|
||||
"node_modules/dbus-next": {
|
||||
"version": "0.9.2",
|
||||
"resolved": "git+ssh://[email protected]/frida/node-dbus-next.git#0e12559b2298c140916541f57fdf9f83cf0b7975",
|
||||
"resolved": "git+ssh://[email protected]/frida-forks/node-dbus-next.git#0e12559b2298c140916541f57fdf9f83cf0b7975",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"@nornagon/put": "0.0.8",
|
||||
@@ -20437,8 +20437,8 @@
|
||||
"peer": true
|
||||
},
|
||||
"dbus-next": {
|
||||
"version": "git+ssh://[email protected]/frida/node-dbus-next.git#0e12559b2298c140916541f57fdf9f83cf0b7975",
|
||||
"from": "dbus-next@frida/node-dbus-next",
|
||||
"version": "git+ssh://[email protected]/frida-forks/node-dbus-next.git#0e12559b2298c140916541f57fdf9f83cf0b7975",
|
||||
"from": "dbus-next@frida-forks/node-dbus-next",
|
||||
"requires": {
|
||||
"@nornagon/put": "0.0.8",
|
||||
"abstract-socket": "^2.0.0",
|
||||
|
||||
@@ -8,7 +8,7 @@
|
||||
},
|
||||
"dependencies": {
|
||||
"core-js": "^3.6.5",
|
||||
"dbus-next": "frida/node-dbus-next",
|
||||
"dbus-next": "frida-forks/node-dbus-next",
|
||||
"duplexify": "^4.1.1",
|
||||
"vue": "^2.6.11",
|
||||
"vue-class-component": "^7.2.3",
|
||||
|
||||
+1
-1
Submodule frida-bindgen updated: df17879d92...eace04901e.
@@ -1,27 +1,3 @@
|
||||
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,
|
||||
@@ -31,7 +7,7 @@ PyEndpointParameters_init (PyEndpointParameters * self,
|
||||
static char * keywords[] = { "address", "port", "certificate", "origin", "auth_service", "asset_root", NULL };
|
||||
char * address = NULL;
|
||||
unsigned short int port = 0;
|
||||
char * certificate_value = NULL;
|
||||
PyObject * certificate_value = NULL;
|
||||
char * origin = NULL;
|
||||
PyObject * auth_service_obj = NULL;
|
||||
char * asset_root_value = NULL;
|
||||
@@ -43,10 +19,10 @@ PyEndpointParameters_init (PyEndpointParameters * self,
|
||||
if (PyGObject_tp_init ((PyObject *) self, args, kw) < 0)
|
||||
return -1;
|
||||
|
||||
if (!PyArg_ParseTupleAndKeywords (args, kw, "|esHesesOes", keywords,
|
||||
if (!PyArg_ParseTupleAndKeywords (args, kw, "|esHOesOes", keywords,
|
||||
"utf-8", &address,
|
||||
&port,
|
||||
"utf-8", &certificate_value,
|
||||
&certificate_value,
|
||||
"utf-8", &origin,
|
||||
&auth_service_obj,
|
||||
"utf-8", &asset_root_value))
|
||||
@@ -74,7 +50,6 @@ beach:
|
||||
|
||||
PyMem_Free (asset_root_value);
|
||||
PyMem_Free (origin);
|
||||
PyMem_Free (certificate_value);
|
||||
PyMem_Free (address);
|
||||
|
||||
return result;
|
||||
|
||||
@@ -910,6 +910,9 @@ propagate_error:
|
||||
static PyObject *
|
||||
PyGObject_marshal_variant (GVariant * variant)
|
||||
{
|
||||
if (variant == NULL)
|
||||
PyFrida_RETURN_NONE;
|
||||
|
||||
switch (g_variant_classify (variant))
|
||||
{
|
||||
case G_VARIANT_CLASS_STRING:
|
||||
@@ -1041,6 +1044,64 @@ PyGObject_marshal_variant_tuple (GVariant * variant)
|
||||
return tuple;
|
||||
}
|
||||
|
||||
static gboolean
|
||||
PyGObject_unmarshal_certificate (PyObject * value,
|
||||
GTlsCertificate ** certificate)
|
||||
{
|
||||
PyObject * bytes;
|
||||
const gchar * str;
|
||||
GError * error = NULL;
|
||||
|
||||
if (value == Py_None)
|
||||
{
|
||||
*certificate = NULL;
|
||||
return TRUE;
|
||||
}
|
||||
|
||||
bytes = PyUnicode_AsUTF8String (value);
|
||||
if (bytes == NULL)
|
||||
return FALSE;
|
||||
str = PyBytes_AsString (bytes);
|
||||
|
||||
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);
|
||||
|
||||
Py_DecRef (bytes);
|
||||
|
||||
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 PyObject *
|
||||
PyGObject_marshal_certificate (GTlsCertificate * certificate)
|
||||
{
|
||||
PyObject * result;
|
||||
gchar * pem;
|
||||
|
||||
if (certificate == NULL)
|
||||
PyFrida_RETURN_NONE;
|
||||
|
||||
g_object_get (certificate, "certificate-pem", &pem, NULL);
|
||||
|
||||
result = PyGObject_marshal_string (pem);
|
||||
|
||||
g_free (pem);
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
static PyObject *
|
||||
PyFrida_raise (GError * error)
|
||||
{
|
||||
|
||||
@@ -35,6 +35,8 @@ 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 gboolean PyGObject_unmarshal_certificate (PyObject * value, GTlsCertificate ** certificate);
|
||||
static PyObject * PyGObject_marshal_certificate (GTlsCertificate * certificate);
|
||||
|
||||
static PyObject * PyFrida_raise (GError * error);
|
||||
static PyObject * PyFrida_marshal_error (GError * error);
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -0,0 +1 @@
|
||||
ScriptExportsAsync = aio._ScriptExports
|
||||
@@ -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
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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"]
|
||||
+416
-53
@@ -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,23 @@ 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, Optional, Tuple, Sequence, Union, TypeVar, cast,
|
||||
overload)
|
||||
|
||||
if sys.version_info >= (3, 11):
|
||||
from typing import NotRequired, ParamSpec, TypedDict
|
||||
else:
|
||||
from typing_extensions import NotRequired, ParamSpec, TypedDict"""
|
||||
|
||||
|
||||
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
|
||||
|
||||
@@ -70,13 +90,29 @@ def _to_envp(value):
|
||||
return value
|
||||
|
||||
|
||||
def _make_options(cls, values):
|
||||
def _make_options(cls, values, selectors):
|
||||
options = cls()
|
||||
for name, value in values.items():
|
||||
setattr(options, name, value)
|
||||
_apply_settings(options, values, selectors)
|
||||
return options
|
||||
|
||||
|
||||
def _apply_settings(impl, values, selectors):
|
||||
for name, value in values.items():
|
||||
if value is None:
|
||||
continue
|
||||
select = selectors.get(name)
|
||||
if select is None:
|
||||
setattr(impl, name, _unwrap(value))
|
||||
elif isinstance(value, dict):
|
||||
add = getattr(impl, select)
|
||||
for key, element in value.items():
|
||||
add(key, _unwrap(element))
|
||||
else:
|
||||
add = getattr(impl, select)
|
||||
for element in value:
|
||||
add(_unwrap(element))
|
||||
|
||||
|
||||
_current_cancellable = threading.local()
|
||||
|
||||
|
||||
@@ -156,14 +192,30 @@ def _to_envp(value):
|
||||
return value
|
||||
|
||||
|
||||
def _make_options(cls, values):
|
||||
def _make_options(cls, values, selectors):
|
||||
options = cls()
|
||||
for name, value in values.items():
|
||||
setattr(options, name, value)
|
||||
_apply_settings(options, values, selectors)
|
||||
return options
|
||||
|
||||
|
||||
_current_cancellable = contextvars.ContextVar("frida_current_cancellable", default=None)
|
||||
def _apply_settings(impl, values, selectors):
|
||||
for name, value in values.items():
|
||||
if value is None:
|
||||
continue
|
||||
select = selectors.get(name)
|
||||
if select is None:
|
||||
setattr(impl, name, _unwrap(value))
|
||||
elif isinstance(value, dict):
|
||||
add = getattr(impl, select)
|
||||
for key, element in value.items():
|
||||
add(key, _unwrap(element))
|
||||
else:
|
||||
add = getattr(impl, select)
|
||||
for element in value:
|
||||
add(_unwrap(element))
|
||||
|
||||
|
||||
_current_cancellable: contextvars.ContextVar = contextvars.ContextVar("frida_current_cancellable", default=None)
|
||||
|
||||
|
||||
def _dispatch(loop, callback, *args):
|
||||
@@ -212,16 +264,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(),
|
||||
"",
|
||||
@@ -246,6 +302,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')
|
||||
@@ -278,14 +337,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(),
|
||||
"",
|
||||
@@ -334,6 +397,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)
|
||||
|
||||
|
||||
@@ -357,7 +422,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}()"""
|
||||
|
||||
|
||||
@@ -371,7 +436,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 = facade_return_type(method, model)
|
||||
|
||||
return f""" async def {method.name}({signature}) -> {ret}:
|
||||
return {call}"""
|
||||
|
||||
|
||||
@@ -379,11 +446,12 @@ def generate_custom_facade_method(
|
||||
method: Method, logic: Union[str, Tuple[str, str]], model: Model, awaitable: bool
|
||||
) -> str:
|
||||
signature = ", ".join(["self"] + method.custom_facade_params)
|
||||
args = facade_call_args(method, model)
|
||||
if method.is_async:
|
||||
invoke = f"_invoke(self._impl.{method.name}, ({method.facade_call_args}))"
|
||||
invoke = f"_invoke(self._impl.{method.name}, ({args}))"
|
||||
call = f"await {invoke}" if awaitable else invoke
|
||||
else:
|
||||
call = f"self._impl.{method.name}({method.facade_call_args})"
|
||||
call = f"self._impl.{method.name}({args})"
|
||||
if method.return_value is not None:
|
||||
call = wrap_result(call, method.return_value.type, model)
|
||||
|
||||
@@ -392,11 +460,27 @@ 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 = facade_return_type(method, model)
|
||||
|
||||
return f""" {keyword} {method.name}({signature}) -> {ret}:
|
||||
{body}
|
||||
return {call}"""
|
||||
|
||||
|
||||
def facade_call_args(method: Method, model: Model) -> str:
|
||||
takes_kwargs = any(param.split(":")[0].strip() == "**kwargs" for param in method.custom_facade_params)
|
||||
args = []
|
||||
for param in method.input_parameters:
|
||||
if param.type.name == "Gio.Cancellable":
|
||||
continue
|
||||
options = resolve_options_type(param.type, model)
|
||||
if options is not None and takes_kwargs:
|
||||
args.append(make_options_expr(options))
|
||||
else:
|
||||
args.append(param.name)
|
||||
return "".join(arg + ", " for arg in args)
|
||||
|
||||
|
||||
def generate_py_exception_reexports(model: Model) -> List[str]:
|
||||
names = [f"{member.js_name}Error" for member in model.error_domain.members]
|
||||
names.append("OperationCancelledError")
|
||||
@@ -425,6 +509,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)
|
||||
|
||||
|
||||
@@ -450,23 +536,62 @@ def generate_facade_init(otype: ObjectType) -> str:
|
||||
{body}"""
|
||||
|
||||
return f""" def __init__(self, *args, **kwargs):
|
||||
self._impl = _frida.{otype.py_name}(
|
||||
*[_unwrap(a) for a in args],
|
||||
**{{k: _unwrap(v) for k, v in kwargs.items()}},
|
||||
)
|
||||
self._impl = _frida.{otype.py_name}(*[_unwrap(a) for a in args])
|
||||
_apply_settings(self._impl, kwargs, {build_option_selectors(otype)})
|
||||
setup = getattr(self, "_setup", None)
|
||||
if setup is not None:
|
||||
setup()"""
|
||||
|
||||
|
||||
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:
|
||||
@@ -500,8 +625,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)
|
||||
@@ -509,7 +635,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
|
||||
@@ -534,19 +660,36 @@ def generate_py_sync_method(method: Method, model: Model) -> Optional[str]:
|
||||
if any(build_sync_param(param) is None for param in method.input_parameters):
|
||||
return None
|
||||
|
||||
signature = ["self"]
|
||||
for param in method.input_parameters:
|
||||
signature.append(f"{param.name}=None" if param.nullable else param.name)
|
||||
names = ", ".join(param.name for param in method.input_parameters)
|
||||
params = [facade_param(param, model) for param in method.input_parameters]
|
||||
names = ", ".join(facade_argument(param, model) 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 = facade_return_type(method, model)
|
||||
|
||||
return f""" def {method.name}({facade_signature(method, params)}) -> {ret}:
|
||||
return {call}"""
|
||||
|
||||
|
||||
def facade_signature(method: Method, params: List[str]) -> str:
|
||||
return ", ".join(["self"] + (method.custom_facade_params or params))
|
||||
|
||||
|
||||
def facade_argument(param, model: Model) -> str:
|
||||
if resolve_input_object_type(param.type, model) is not None:
|
||||
return f"_unwrap({param.name})"
|
||||
return param.name
|
||||
|
||||
|
||||
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
|
||||
@@ -560,12 +703,14 @@ 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 = facade_return_type(method, model)
|
||||
|
||||
return f""" def {method.name}({signature}) -> {ret}:
|
||||
return {call}"""
|
||||
|
||||
|
||||
def build_facade_async_parts(method: Method, model: Model) -> Optional[Tuple[str, str]]:
|
||||
signature = ["self"]
|
||||
params = []
|
||||
args = []
|
||||
for param in method.input_parameters:
|
||||
if param.type.name == "Gio.Cancellable":
|
||||
@@ -574,20 +719,35 @@ def build_facade_async_parts(method: Method, model: Model) -> Optional[Tuple[str
|
||||
return None
|
||||
options = resolve_options_type(param.type, model)
|
||||
if options is not None:
|
||||
signature.append("**kwargs")
|
||||
args.append(f"_make_options(_frida.{options.py_name}, kwargs)")
|
||||
params.append("**kwargs")
|
||||
args.append(make_options_expr(options))
|
||||
elif resolve_input_object_type(param.type, model) is not None:
|
||||
signature.append(f"{param.name}=None")
|
||||
params.append(f"{param.name}: Optional[{pyi_type(param.type, model)}] = None")
|
||||
args.append(f"_unwrap({param.name})")
|
||||
else:
|
||||
signature.append(param.name)
|
||||
params.append(facade_param(param, model))
|
||||
args.append(param.name)
|
||||
|
||||
if method.return_value is not None:
|
||||
if build_return_marshal(method.return_value.type, model) is None:
|
||||
return None
|
||||
|
||||
return ", ".join(signature), "".join(arg + ", " for arg in args)
|
||||
return facade_signature(method, params), "".join(arg + ", " for arg in args)
|
||||
|
||||
|
||||
def make_options_expr(options: ObjectType) -> str:
|
||||
return f"_make_options(_frida.{options.py_name}, kwargs, {build_option_selectors(options)})"
|
||||
|
||||
|
||||
def build_option_selectors(options: ObjectType) -> str:
|
||||
selectors = {}
|
||||
otype = options
|
||||
while otype is not None:
|
||||
for method in otype.methods:
|
||||
if method.is_select_method:
|
||||
selectors[method.select_plural_noun] = method.name
|
||||
otype = otype.parent
|
||||
return repr(selectors)
|
||||
|
||||
|
||||
def wrap_result(call: str, type: Type, model: Model) -> str:
|
||||
@@ -618,6 +778,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)
|
||||
|
||||
@@ -628,13 +793,22 @@ def generate_pyi_class(otype: ObjectType, model: Model) -> List[str]:
|
||||
|
||||
body = []
|
||||
|
||||
ctor = next(iter(otype.constructors), None)
|
||||
if ctor is not None and not ctor.throws:
|
||||
params = pyi_parameters(ctor.input_parameters, model)
|
||||
primary = inline_constructor(otype)
|
||||
if primary is not None:
|
||||
params = pyi_parameters(primary.input_parameters, model)
|
||||
if params is not None:
|
||||
signature = ", ".join(["self"] + params)
|
||||
body.append(f" def __init__({signature}) -> None: ...")
|
||||
|
||||
for ctor in otype.constructors:
|
||||
if ctor is primary:
|
||||
continue
|
||||
params = pyi_parameters(ctor.input_parameters, model)
|
||||
if params is None:
|
||||
continue
|
||||
body.append(" @staticmethod")
|
||||
body.append(f" def {ctor.name}({', '.join(params)}) -> {otype.py_name}: ...")
|
||||
|
||||
if otype.name == "Object":
|
||||
body.append(" def on(self, signal: str, callback: Callable[..., Any]) -> None: ...")
|
||||
body.append(" def off(self, signal: str, callback: Callable[..., Any]) -> None: ...")
|
||||
@@ -643,8 +817,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)
|
||||
@@ -690,7 +869,13 @@ def pyi_method(method: Method, model: Model) -> Optional[str]:
|
||||
if params is None:
|
||||
return None
|
||||
signature = ", ".join(["self"] + params)
|
||||
ret = "None" if method.return_value is None else pyi_type(method.return_value.type, model)
|
||||
out = method.optional_out_parameter
|
||||
if out is not None:
|
||||
ret = f"Optional[{pyi_type(out.type, model)}]"
|
||||
elif method.return_value is None:
|
||||
ret = "None"
|
||||
else:
|
||||
ret = pyi_type(method.return_value.type, model)
|
||||
return f" def {method.name}({signature}) -> {ret}: ..."
|
||||
|
||||
|
||||
@@ -735,6 +920,8 @@ def pyi_type(type: Type, model: Model) -> str:
|
||||
return "dict"
|
||||
if name == "GLib.Variant":
|
||||
return "Any"
|
||||
if name == "Gio.TlsCertificate":
|
||||
return "str"
|
||||
if resolve_enumeration(type, model) is not None:
|
||||
return "str"
|
||||
obj = resolve_object_type(type, model)
|
||||
@@ -776,6 +963,10 @@ def generate_extension_c(model: Model) -> str:
|
||||
code += generate_interface_implementation(otype)
|
||||
|
||||
code += generate_object_type_constructor(otype)
|
||||
for ctor in factory_constructors(otype):
|
||||
marshals = build_constructor_marshals(ctor)
|
||||
assert marshals is not None
|
||||
code += generate_factory_function(otype, ctor, marshals)
|
||||
|
||||
return code
|
||||
|
||||
@@ -1050,6 +1241,10 @@ def generate_prototypes(model: Model) -> str:
|
||||
"",
|
||||
f"static int {otype_cprefix}_init ({otype_cprefix} * self, PyObject * args, PyObject * kw);",
|
||||
]
|
||||
for ctor in factory_constructors(otype):
|
||||
prototypes.append(
|
||||
f"static PyObject * {otype_cprefix}_{ctor.name} (PyObject * self, PyObject * args, PyObject * kw);"
|
||||
)
|
||||
if implementable_interface(otype):
|
||||
prototypes += [
|
||||
f"static int {otype_cprefix}_traverse (PyObject * self, visitproc visit, void * arg);",
|
||||
@@ -1166,6 +1361,9 @@ def generate_object_type_methods(otype: ObjectType) -> str:
|
||||
functions.append(function)
|
||||
entries.append(entry)
|
||||
|
||||
for ctor in factory_constructors(otype):
|
||||
entries.append(factory_method_entry(otype, ctor))
|
||||
|
||||
return f"""{"".join(functions)}
|
||||
static PyMethodDef {otype.c_symbol_prefix}_methods[] =
|
||||
{{
|
||||
@@ -1181,6 +1379,11 @@ def generate_method(otype: ObjectType, method: Method) -> Optional[Tuple[str, st
|
||||
marshal = synchronous_method_return_marshal(method)
|
||||
if marshal is None:
|
||||
return None
|
||||
out = method.optional_out_parameter
|
||||
if out is not None:
|
||||
marshal = build_return_marshal(out.type, method.object_type.model)
|
||||
if marshal is None:
|
||||
return None
|
||||
params = []
|
||||
for param in method.input_parameters:
|
||||
sync_param = build_sync_param(param)
|
||||
@@ -1524,14 +1727,17 @@ def generate_synchronous_method(otype: ObjectType, method: Method, marshal: str,
|
||||
indent = " " * (len(otype.c_symbol_prefix) + len(method.name) + 3)
|
||||
handle = f"({otype.c_type} *) PY_GOBJECT_HANDLE (self)"
|
||||
returns_strv = method.return_value is not None and method.return_value.type.name == "utf8[]"
|
||||
out = method.optional_out_parameter
|
||||
call_arg_list = [handle] + [p.call_arg for p in params]
|
||||
if returns_strv:
|
||||
call_arg_list.append("&retval_length")
|
||||
if out is not None:
|
||||
call_arg_list.append(f"&{out.name}")
|
||||
if method.throws:
|
||||
call_arg_list.append("&error")
|
||||
call = f"{method.c_identifier} ({', '.join(call_arg_list)})"
|
||||
|
||||
if not params and not method.throws and not returns_strv:
|
||||
if not params and not method.throws and not returns_strv and out is None:
|
||||
if method.return_value is None:
|
||||
body = f" {call};\n\n PyFrida_RETURN_NONE;"
|
||||
else:
|
||||
@@ -1550,6 +1756,8 @@ static PyObject *
|
||||
decls.append(" GError * error = NULL;")
|
||||
if returns_strv:
|
||||
decls.append(" gint retval_length;")
|
||||
if out is not None:
|
||||
decls.append(f" {out_parameter_storage_type(out)} {out.name};")
|
||||
for p in params:
|
||||
decls += [f" {line}" for line in p.decl.splitlines()]
|
||||
|
||||
@@ -1564,7 +1772,15 @@ static PyObject *
|
||||
|
||||
pre = "".join(f"\n {p.pre}\n" for p in params if p.pre)
|
||||
|
||||
if method.return_value is None:
|
||||
if out is not None:
|
||||
invocation = f" gboolean found = {call};"
|
||||
success = (
|
||||
" if (found)\n"
|
||||
f" result = {marshal.format(value=out.name)};\n"
|
||||
" else\n"
|
||||
" {\n Py_IncRef (Py_None);\n result = Py_None;\n }"
|
||||
)
|
||||
elif method.return_value is None:
|
||||
invocation = f" {call};"
|
||||
success = " Py_IncRef (Py_None);\n result = Py_None;"
|
||||
elif returns_strv:
|
||||
@@ -1609,6 +1825,22 @@ static PyObject *
|
||||
"""
|
||||
|
||||
|
||||
def facade_return_type(method: Method, model: Model) -> str:
|
||||
out = method.optional_out_parameter
|
||||
if out is not None:
|
||||
return f"Optional[{pyi_type(out.type, model)}]"
|
||||
if method.return_value is None:
|
||||
return "None"
|
||||
return pyi_type(method.return_value.type, model)
|
||||
|
||||
|
||||
def out_parameter_storage_type(param: Parameter) -> str:
|
||||
"""What the caller declares: the out parameter's own type, less the pointer."""
|
||||
c = param.type.c
|
||||
assert c.endswith("*"), c
|
||||
return c[:-1].strip()
|
||||
|
||||
|
||||
def synchronous_method_return_marshal(method: Method) -> Optional[str]:
|
||||
if method.is_async or method.is_property_accessor:
|
||||
return None
|
||||
@@ -1691,6 +1923,16 @@ def build_sync_param(param: Parameter) -> Optional["SyncParam"]:
|
||||
cleanup=f"g_clear_pointer (&{name}, g_variant_unref);",
|
||||
)
|
||||
|
||||
if resolve_input_object_type(param.type, param.object_type.model) is not None:
|
||||
return SyncParam(
|
||||
decl=f"PyObject * {name}_obj;\n{param.type.c} {name} = NULL;",
|
||||
fmt="O",
|
||||
parse_args=[f"&{name}_obj"],
|
||||
pre=f"""if ({name}_obj != Py_None)
|
||||
{name} = ({param.type.c}) PY_GOBJECT_HANDLE ({name}_obj);""",
|
||||
call_arg=name,
|
||||
)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
@@ -1869,6 +2111,17 @@ def build_setter_body(param: Parameter, set_call) -> Optional[str]:
|
||||
g_hash_table_unref (dict);
|
||||
"""
|
||||
|
||||
if type_name == "Gio.TlsCertificate":
|
||||
return f""" GTlsCertificate * certificate;
|
||||
|
||||
if (!PyGObject_unmarshal_certificate (value, &certificate))
|
||||
return -1;
|
||||
|
||||
{set_call("certificate")}
|
||||
|
||||
g_clear_object (&certificate);
|
||||
"""
|
||||
|
||||
if resolve_input_object_type(param.type, param.object_type.model) is not None:
|
||||
handle = f"({param.type.c}) ((value != Py_None) ? PY_GOBJECT_HANDLE (value) : NULL)"
|
||||
return f" {set_call(handle)}\n"
|
||||
@@ -1948,6 +2201,8 @@ def build_return_marshal(type: Type, model: Model) -> Optional[str]:
|
||||
return "PyGObject_marshal_vardict ({value})"
|
||||
if name == "GLib.Variant":
|
||||
return "PyGObject_marshal_variant ({value})"
|
||||
if name == "Gio.TlsCertificate":
|
||||
return "PyGObject_marshal_certificate ({value})"
|
||||
|
||||
enum = resolve_enumeration(type, model)
|
||||
if enum is not None:
|
||||
@@ -2094,28 +2349,55 @@ def generate_object_type_constructor(otype: ObjectType) -> str:
|
||||
if implementable_interface(otype):
|
||||
return generate_interface_init(otype)
|
||||
|
||||
ctor = next(iter(otype.constructors), None)
|
||||
ctor = primary_constructor(otype)
|
||||
|
||||
if ctor is None:
|
||||
return generate_bare_init(otype)
|
||||
|
||||
marshals: Optional[List[ParamMarshal]] = None
|
||||
if not ctor.throws:
|
||||
marshals = []
|
||||
for param in ctor.input_parameters:
|
||||
marshal = build_param_marshal(param)
|
||||
if marshal is None:
|
||||
marshals = None
|
||||
break
|
||||
marshal.optional = param.optional
|
||||
marshals.append(marshal)
|
||||
if ctor.throws:
|
||||
return generate_unconstructable_init(otype)
|
||||
|
||||
marshals = build_constructor_marshals(ctor)
|
||||
if marshals is None:
|
||||
return generate_unconstructable_init(otype)
|
||||
|
||||
return generate_constructor_init(otype, ctor, marshals)
|
||||
|
||||
|
||||
def primary_constructor(otype: ObjectType) -> Optional[Procedure]:
|
||||
if not otype.constructors:
|
||||
return None
|
||||
return next((c for c in otype.constructors if c.name == "new"), otype.constructors[0])
|
||||
|
||||
|
||||
def inline_constructor(otype: ObjectType) -> Optional[Procedure]:
|
||||
"""The constructor __init__ takes; one that throws is reached as a factory instead."""
|
||||
ctor = primary_constructor(otype)
|
||||
if ctor is None or ctor.throws:
|
||||
return None
|
||||
return ctor
|
||||
|
||||
|
||||
def factory_constructors(otype: ObjectType) -> List[Procedure]:
|
||||
primary = inline_constructor(otype)
|
||||
return [
|
||||
ctor
|
||||
for ctor in otype.constructors
|
||||
if ctor is not primary and not ctor.is_async and build_constructor_marshals(ctor) is not None
|
||||
]
|
||||
|
||||
|
||||
def build_constructor_marshals(ctor: Procedure) -> Optional[List["ParamMarshal"]]:
|
||||
marshals = []
|
||||
for param in ctor.input_parameters:
|
||||
marshal = build_param_marshal(param)
|
||||
if marshal is None:
|
||||
return None
|
||||
marshal.optional = param.optional
|
||||
marshals.append(marshal)
|
||||
return marshals
|
||||
|
||||
|
||||
def generate_interface_init(otype: ObjectType) -> str:
|
||||
n = interface_impl_names(otype)
|
||||
indent = constructor_indent(otype)
|
||||
@@ -2216,7 +2498,88 @@ static int
|
||||
{generate_constructor_tail(marshals)}"""
|
||||
|
||||
|
||||
def generate_argument_parse(marshals: List["ParamMarshal"]) -> str:
|
||||
def factory_method_entry(otype: ObjectType, ctor: Procedure) -> str:
|
||||
return (
|
||||
f' {{ "{ctor.name}", (PyCFunction) {otype.c_symbol_prefix}_{ctor.name}, '
|
||||
"METH_VARARGS | METH_KEYWORDS | METH_STATIC, NULL },"
|
||||
)
|
||||
|
||||
|
||||
def generate_factory_function(otype: ObjectType, ctor: Procedure, marshals: List["ParamMarshal"]) -> str:
|
||||
indent = " " * (len(otype.c_symbol_prefix) + len(ctor.name) + 3)
|
||||
|
||||
decls = [line for m in marshals for line in m.decls]
|
||||
decls.append(f"{otype.c_type} * handle;")
|
||||
if ctor.throws:
|
||||
decls.append("GError * error = NULL;")
|
||||
decls.append("PyObject * result = NULL;")
|
||||
decls_block = indent_c_code("\n".join(decls), 1, prologue="\n")
|
||||
|
||||
keyword_decl = ""
|
||||
parse_block = ""
|
||||
if marshals:
|
||||
keywords = ", ".join([f'"{m.keyword}"' for m in marshals] + ["NULL"])
|
||||
keyword_decl = f"\n static char * keywords[] = {{ {keywords} }};"
|
||||
parse_block = "\n" + generate_argument_parse(marshals, on_failure="return NULL;")
|
||||
|
||||
post_block = ""
|
||||
for m in marshals:
|
||||
if m.post is not None:
|
||||
post_block += indent_c_code(m.post, 1, prologue="\n") + "\n"
|
||||
|
||||
call_args = [m.call_arg for m in marshals]
|
||||
if ctor.throws:
|
||||
call_args.append("&error")
|
||||
call = f"handle = ({otype.c_type} *) {ctor.c_identifier} ({', '.join(call_args)});"
|
||||
|
||||
error_block = ""
|
||||
if ctor.throws:
|
||||
error_block = """
|
||||
if (error != NULL)
|
||||
{
|
||||
result = PyFrida_raise (error);
|
||||
goto beach;
|
||||
}
|
||||
"""
|
||||
|
||||
take_handle = f"result = PyGObject_new_take_handle (g_steal_pointer (&handle), PYFRIDA_TYPE ({otype.py_name}));"
|
||||
|
||||
return f"""
|
||||
static PyObject *
|
||||
{otype.c_symbol_prefix}_{ctor.name} (PyObject * self,
|
||||
{indent}PyObject * args,
|
||||
{indent}PyObject * kw)
|
||||
{{{keyword_decl}{decls_block}
|
||||
{parse_block}{post_block}
|
||||
{call}
|
||||
{error_block}
|
||||
{take_handle}
|
||||
|
||||
{generate_factory_tail(marshals, ctor.throws)}"""
|
||||
|
||||
|
||||
def generate_factory_tail(marshals: List["ParamMarshal"], throws: bool) -> str:
|
||||
cleanups = [m.cleanup for m in marshals if m.cleanup is not None]
|
||||
cleanup_block = indent_c_code("\n".join(reversed(cleanups)), 1) if cleanups else ""
|
||||
|
||||
if throws or constructor_needs_beach(marshals):
|
||||
prologue = f"{cleanup_block}\n\n" if cleanup_block else ""
|
||||
return f"""beach:
|
||||
{prologue} return result;
|
||||
}}
|
||||
"""
|
||||
if cleanup_block:
|
||||
return f"""{cleanup_block}
|
||||
|
||||
return result;
|
||||
}}
|
||||
"""
|
||||
return """ return result;
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
def generate_argument_parse(marshals: List["ParamMarshal"], on_failure: str = "return -1;") -> str:
|
||||
fmt = ""
|
||||
optional_started = False
|
||||
for m in marshals:
|
||||
@@ -2230,7 +2593,7 @@ def generate_argument_parse(marshals: List["ParamMarshal"]) -> str:
|
||||
indent = " " * len(" if (!PyArg_ParseTupleAndKeywords (")
|
||||
parse_args_str = (",\n" + indent).join(parse_args)
|
||||
return f""" if (!PyArg_ParseTupleAndKeywords ({parse_args_str}))
|
||||
return -1;
|
||||
{on_failure}
|
||||
"""
|
||||
|
||||
|
||||
|
||||
@@ -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,23 +104,22 @@ 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"],
|
||||
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)",
|
||||
),
|
||||
param_typings=["target: ProcessTarget", "**kwargs: Any"],
|
||||
custom_logic=_RESOLVE_PID,
|
||||
),
|
||||
},
|
||||
custom_code=CustomCode(
|
||||
@@ -137,9 +134,10 @@ if aux:
|
||||
),
|
||||
methods={
|
||||
"post": MethodCustomizations(
|
||||
param_typings=["message", "data=None"],
|
||||
param_typings=["message: Any", "data: Optional[bytes] = None"],
|
||||
custom_logic="json = _to_json(message)",
|
||||
),
|
||||
"enable_debugger": MethodCustomizations(param_typings=["port: int = 0"]),
|
||||
"is_destroyed": MethodCustomizations(as_property=True),
|
||||
},
|
||||
),
|
||||
@@ -147,7 +145,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:
|
||||
@@ -158,7 +161,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()
|
||||
@@ -170,7 +178,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:
|
||||
@@ -179,7 +187,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:
|
||||
@@ -198,13 +210,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
|
||||
@@ -217,7 +229,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:
|
||||
@@ -225,7 +237,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:
|
||||
@@ -381,4 +393,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",),
|
||||
)
|
||||
@@ -17,6 +17,10 @@ class Model(core.Model):
|
||||
|
||||
|
||||
class ObjectType(core.ObjectType):
|
||||
@cached_property
|
||||
def methods(self) -> List["Method"]:
|
||||
return [m for m in super().methods if not m.out_parameters or m.optional_out_parameter is not None]
|
||||
|
||||
@cached_property
|
||||
def py_name(self) -> str:
|
||||
custom = self.customizations
|
||||
@@ -107,6 +111,14 @@ class Method(core.Method):
|
||||
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
|
||||
@@ -136,10 +148,6 @@ class Method(core.Method):
|
||||
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
|
||||
@@ -192,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
|
||||
@@ -271,6 +281,7 @@ def _make_class(
|
||||
type_struct,
|
||||
parent,
|
||||
constructors,
|
||||
functions,
|
||||
methods,
|
||||
properties,
|
||||
signals,
|
||||
@@ -285,6 +296,7 @@ def _make_class(
|
||||
type_struct,
|
||||
parent,
|
||||
constructors,
|
||||
functions,
|
||||
methods,
|
||||
properties,
|
||||
signals,
|
||||
@@ -301,6 +313,7 @@ def _make_interface(
|
||||
type_struct,
|
||||
parent,
|
||||
constructors,
|
||||
functions,
|
||||
methods,
|
||||
properties,
|
||||
signals,
|
||||
@@ -314,6 +327,7 @@ def _make_interface(
|
||||
type_struct,
|
||||
parent,
|
||||
constructors,
|
||||
functions,
|
||||
methods,
|
||||
properties,
|
||||
signals,
|
||||
|
||||
+1
-1
Submodule releng updated: abb8238970...ce275018bb.
@@ -1,6 +1,6 @@
|
||||
[wrap-git]
|
||||
url = https://github.com/frida/frida-core.git
|
||||
revision = 17.16.2
|
||||
revision = 17.26.0
|
||||
depth = 1
|
||||
|
||||
[provide]
|
||||
|
||||
@@ -17,6 +17,27 @@ 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"
|
||||
CERTIFICATE = """\
|
||||
-----BEGIN CERTIFICATE-----
|
||||
MIIDGzCCAgOgAwIBAgIUEoCZX+FRWNTzcvvrCE2chFvE0EgwDQYJKoZIhvcNAQEL
|
||||
BQAwHDEaMBgGA1UEAwwRZnJpZGEtcHl0aG9uLXRlc3QwIBcNMjYwNzMwMjIyNjAy
|
||||
WhgPMjEyNjA3MDYyMjI2MDJaMBwxGjAYBgNVBAMMEWZyaWRhLXB5dGhvbi10ZXN0
|
||||
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAtKTIUbHLhkqRKdKqXuMc
|
||||
u7q8OZD2eKdykp4Qvv+I5JcmNW7nlW+58zTCLaqUCsSrojDb8T2qvtRJbaJYAofE
|
||||
P7bhb5ROOVwsyOjhvP7rZ4zRXIDAUlB7zE15yNkWL+Rh4l3wRxMOMIB2HQlurtDk
|
||||
PIqPOITwRvrADKZt/73iMASwrkkqKnl/F3rLT0JQrd907sBdGGZNbJM9L4TUBT9t
|
||||
MBBl6qXtDEFyBfeoqlnrLUTN/Jwyka+/6n6eZCN8g2g5tS1Lc7LFjX4iBG9ibgaH
|
||||
hrNSTM+haLImFS3dtXFr4rOCAHnDYYvkCY1nuZOXannCvZowv0xczqpGLbj5IYJK
|
||||
MQIDAQABo1MwUTAdBgNVHQ4EFgQUi83cV5yMTdWaVeRpgNO3vyeitMMwHwYDVR0j
|
||||
BBgwFoAUi83cV5yMTdWaVeRpgNO3vyeitMMwDwYDVR0TAQH/BAUwAwEB/zANBgkq
|
||||
hkiG9w0BAQsFAAOCAQEADScSFuasB4mgOb/kktq3O09rf/uUIa6OEjdFRec/i2+E
|
||||
ZAUd5onsWQTHm4yhjqQwqVBte58tTqZB7S1HMCy5sEgEVUFnTrMUqtrf/WAnsWA6
|
||||
SdjNr7h/aRnLHHysLkbl0Zo4fOIWIYSJpLn39e0OzcNBjEHN5U9HE/m/2LFku1Kb
|
||||
KMTYXrhBKvVPpZLxZyoLz00EM7euaNx3juAgfQdSDfKGahf5CqqMJ/VmITISDJDR
|
||||
jtN3wcv0/nZA5x/b73QCj0jqIEWqguYf4GLWitxGsSdT2dBEpT88Kz3Ibwb8uRT+
|
||||
Ls3eRsdoqxiOT0E7Z44O8gjbqsYOr/jdMShwEytDwQ==
|
||||
-----END CERTIFICATE-----
|
||||
"""
|
||||
|
||||
|
||||
@unittest.skipUnless(
|
||||
@@ -131,6 +152,29 @@ class TestExtension(unittest.TestCase):
|
||||
options.argv = None
|
||||
self.assertIsNone(options.argv)
|
||||
|
||||
def test_certificate_property_accepts_pem_and_path(self):
|
||||
import tempfile
|
||||
|
||||
options = self._frida.RemoteDeviceOptions()
|
||||
self.assertIsNone(options.certificate)
|
||||
|
||||
options.certificate = CERTIFICATE
|
||||
self.assertEqual(options.certificate, CERTIFICATE)
|
||||
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".pem") as f:
|
||||
f.write(CERTIFICATE)
|
||||
f.flush()
|
||||
options.certificate = f.name
|
||||
self.assertEqual(options.certificate, CERTIFICATE)
|
||||
|
||||
options.certificate = None
|
||||
self.assertIsNone(options.certificate)
|
||||
|
||||
def test_certificate_property_rejects_missing_file(self):
|
||||
options = self._frida.RemoteDeviceOptions()
|
||||
with self.assertRaises(self._frida.InvalidArgumentError):
|
||||
options.certificate = "/nonexistent.pem"
|
||||
|
||||
def test_vardict_property_round_trips(self):
|
||||
options = self._frida.SpawnOptions()
|
||||
options.aux = {"name": "value", "count": 42, "flag": True}
|
||||
@@ -646,6 +690,16 @@ class TestFacade(unittest.TestCase):
|
||||
finally:
|
||||
session.detach()
|
||||
|
||||
def test_attach_accepts_linker_notifier_offsets(self):
|
||||
session = self.frida.get_local_device().attach(0, linker_notifier_offsets=[0x100])
|
||||
session.detach()
|
||||
|
||||
def test_enable_debugger_defaults_to_any_port(self):
|
||||
import inspect
|
||||
|
||||
for script in [self.frida.Script, self.frida.aio.Script]:
|
||||
self.assertEqual(inspect.signature(script.enable_debugger).parameters["port"].default, 0)
|
||||
|
||||
def test_file_monitor_reports_changes(self):
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
Reference in new issue
Block a user