Compare commits

...
112 Commits
Author SHA1 Message Date
Ole André Vadla Ravnås 735d34d175 subprojects: Prepare for release 2026-10-09 14:19:26 +02:00
Ole André Vadla Ravnås a6feef0770 subprojects: Bump outdated 2026-10-09 14:16:52 +02:00
Ole André Vadla Ravnås 418c5e3f43 subprojects: Prepare for release 2026-10-08 14:49:02 +02:00
Ole André Vadla Ravnås a1929a4dc5 subprojects: Bump outdated 2026-10-08 14:32:45 +02:00
Ole André Vadla Ravnås aacd1913e0 subprojects: Prepare for release 2026-10-07 14:27:03 +02:00
Ole André Vadla Ravnås ea6c66da96 subprojects: Bump outdated 2026-10-07 14:24:00 +02:00
Ole André Vadla Ravnås 57b6cbae38 submodules: Bump releng 2026-10-07 14:24:00 +02:00
Ole André Vadla Ravnås 1b18f0174d subprojects: Prepare for release 2026-10-05 10:00:33 +02:00
Ole André Vadla Ravnås 7f04d40687 subprojects: Bump outdated 2026-10-05 09:57:11 +02:00
Ole André Vadla Ravnås c69db7eb40 submodules: Bump releng 2026-10-05 09:57:10 +02:00
Ole André Vadla Ravnås ce49a3102a subprojects: Prepare for release 2026-10-04 11:35:34 +02:00
Ole André Vadla Ravnås 288133374a subprojects: Bump outdated 2026-10-04 11:34:54 +02:00
Ole André Vadla Ravnås 56843bb3f8 subprojects: Prepare for release 2026-10-03 07:56:17 +02:00
Ole André Vadla Ravnås d90f991948 subprojects: Bump outdated 2026-10-03 07:55:28 +02:00
Ole André Vadla Ravnås c744c2a12a subprojects: Prepare for release 2026-10-02 20:28:27 +02:00
Ole André Vadla Ravnås 48631baafa subprojects: Bump outdated 2026-10-02 20:27:05 +02:00
Ole André Vadla Ravnås c711f8f040 subprojects: Prepare for release 2026-10-02 01:36:43 +02:00
Ole André Vadla Ravnås 049674da0a subprojects: Bump outdated 2026-10-02 01:32:18 +02:00
Ole André Vadla Ravnås e1c24e8187 bindgen: Marshal null variants as None 2026-10-01 18:24:45 +02:00
Ole André Vadla Ravnås 9947655e6a subprojects: Prepare for release 2026-09-25 13:15:47 +02:00
Ole André Vadla Ravnås e5b15dce63 subprojects: Bump outdated 2026-09-25 13:14:11 +02:00
Ole André Vadla Ravnås e846eb2815 subprojects: Bump outdated 2026-09-25 12:18:59 +02:00
Ole André Vadla Ravnås bc685fd8c1 subprojects: Bump outdated 2026-09-24 21:35:04 +02:00
Ole André Vadla Ravnås bcaa1a099e subprojects: Bump outdated 2026-09-23 00:02:59 +02:00
Ole André Vadla Ravnås 4e782ccbdf submodules: Bump releng 2026-09-23 00:02:58 +02:00
Ole André Vadla Ravnås 9823368745 subprojects: Bump outdated 2026-09-22 23:47:25 +02:00
Ole André Vadla Ravnås c6e5a8d4d9 submodules: Bump releng 2026-09-22 23:47:25 +02:00
Ole André Vadla Ravnås eb15035950 subprojects: Bump outdated 2026-09-22 21:04:23 +02:00
Ole André Vadla Ravnås ec509287e1 subprojects: Bump outdated 2026-09-22 15:57:26 +02:00
Ole André Vadla Ravnås 0c52c00bad submodules: Bump releng 2026-09-22 15:57:26 +02:00
Ole André Vadla Ravnås 8f31efacf2 Format with the current Black
Black has changed how it lays out a call whose argument is a
multiline string, and a `with` statement holding several contexts.
2026-09-17 19:01:25 +02:00
Ole André Vadla Ravnås 01fd680353 bindgen: Fix dropped constructors and out params
A static function that makes the class it sits on was not read at all,
so every named constructor in the API was missing. One that throws was
then missing twice over: skipped for __init__ because it throws, and
skipped as a factory for being the primary, leaving nothing to make
the class with.

An out parameter was dropped along with the value it carried, so a
`gboolean` method that answers through one now binds as Optional, and
any other out parameter is refused rather than bound as a result the
caller cannot read.
2026-09-17 18:54:58 +02:00
Ole André Vadla Ravnås 00e6231b89 subprojects: Bump outdated 2026-09-16 19:28:53 +02:00
Ole André Vadla Ravnås aee40b044d subprojects: Prepare for release 2026-09-09 20:37:20 +02:00
Ole André Vadla Ravnås b8c22c18f1 subprojects: Bump outdated 2026-09-09 20:33:13 +02:00
Ole André Vadla Ravnås b0a080cd14 subprojects: Bump outdated 2026-09-09 20:16:39 +02:00
Ole André Vadla Ravnås e071ead36d subprojects: Bump outdated 2026-09-09 00:52:10 +02:00
Ole André Vadla Ravnås 2ff7776289 submodules: Bump releng 2026-09-09 00:52:10 +02:00
Ole André Vadla Ravnås df1b3af738 subprojects: Bump outdated 2026-09-09 00:47:02 +02:00
Ole André Vadla Ravnås 03d3b8ab0c subprojects: Bump outdated 2026-09-08 22:34:08 +02:00
Ole André Vadla Ravnås dd0639b6a0 submodules: Bump releng 2026-09-08 22:34:08 +02:00
Ole André Vadla Ravnås bfc6d48b72 bindgen: Fix typing imports on Python < 3.11
NotRequired, ParamSpec and TypedDict come from typing_extensions
there.
2026-09-08 14:14:07 +02:00
Ole André Vadla Ravnås 903143cfd2 subprojects: Bump outdated 2026-09-08 12:14:50 +02:00
Ole André Vadla Ravnås c91bd2d443 subprojects: Bump outdated 2026-09-08 11:39:29 +02:00
Ole André Vadla Ravnås c918c1e516 subprojects: Bump outdated 2026-09-07 17:39:36 +02:00
Ole André Vadla Ravnås ec645f2118 subprojects: Bump outdated 2026-09-07 11:47:57 +02:00
Ole André Vadla Ravnås 41a658af3f bindgen: Blacken model.py 2026-09-06 12:14:30 +02:00
Ole André Vadla Ravnås 9a86bd589b bindgen: Expose named constructors as factories
Emit them all rather than the first alone, and marshal a bytes
argument through GBytes. Keep new as the primary constructor.
2026-09-06 12:04:31 +02:00
Ole André Vadla Ravnås 82293f6c56 subprojects: Bump outdated 2026-09-06 12:02:40 +02:00
Ole André Vadla Ravnås b5c8757514 subprojects: Bump outdated 2026-09-04 17:27:46 +02:00
Ole André Vadla Ravnås 8c7b77cf52 subprojects: Bump outdated 2026-09-04 12:26:26 +02:00
Ole André Vadla Ravnås bbb129ac8d subprojects: Bump outdated 2026-09-03 22:40:11 +02:00
Ole André Vadla Ravnås 51c411d3f4 subprojects: Bump outdated 2026-09-03 18:19:36 +02:00
Ole André Vadla Ravnås a0dd2b7463 subprojects: Bump outdated 2026-09-03 14:43:19 +02:00
Ole André Vadla Ravnås fcc203f604 subprojects: Bump outdated 2026-09-02 18:42:33 +02:00
Ole André Vadla Ravnås a04b4a328e submodules: Bump releng 2026-09-02 18:42:33 +02:00
Ole André Vadla Ravnås f4109be14e subprojects: Bump outdated 2026-09-02 00:30:41 +02:00
Ole André Vadla Ravnås 786398a682 submodules: Bump releng 2026-09-02 00:30:41 +02:00
Ole André Vadla Ravnås 738060a8c5 subprojects: Bump outdated 2026-09-01 21:38:50 +02:00
Ole André Vadla Ravnås b5fb9987fc submodules: Bump releng 2026-09-01 21:38:50 +02:00
Ole André Vadla Ravnås b69d19f673 subprojects: Bump outdated 2026-09-01 15:16:05 +02:00
Ole André Vadla Ravnås 6622f997cd submodules: Bump releng 2026-09-01 15:16:05 +02:00
Ole André Vadla Ravnås 6eeeae51c8 subprojects: Bump outdated 2026-09-01 14:05:16 +02:00
Ole André Vadla Ravnås 679805bc8c submodules: Bump releng 2026-09-01 14:05:16 +02:00
Ole André Vadla Ravnås ec58231298 subprojects: Bump outdated 2026-08-27 18:26:09 +02:00
Ole André Vadla Ravnås 0caa5d4cf6 submodules: Bump releng 2026-08-27 18:26:09 +02:00
Ole André Vadla Ravnås 97d79c14a3 subprojects: Bump outdated 2026-08-18 15:32:21 +02:00
Ole André Vadla Ravnås e46178a118 submodules: Bump releng 2026-08-18 15:32:21 +02:00
Ole André Vadla Ravnås 1e631075e4 subprojects: Bump outdated 2026-08-18 12:50:57 +02:00
Ole André Vadla Ravnås 9cd85d0cfc subprojects: Bump outdated 2026-08-18 00:13:05 +02:00
Ole André Vadla Ravnås 4baee2713e subprojects: Bump outdated 2026-08-17 21:10:56 +02:00
Ole André Vadla Ravnås 4f33ee3945 submodules: Bump releng 2026-08-17 21:10:56 +02:00
Ole André Vadla Ravnås 6f890ee8d4 Take object args and settings kwargs
Sync methods dropped parameters of object type, so PeerOptions.add_relay
never reached the extension even though the facade referred to it.

Constructor kwargs were forwarded to an extension init that ignores them,
leaving them silently unapplied. Apply them to the impl instead, reusing
the options selectors so adders take a list, or a dict when they take a
name and a value.
2026-08-17 20:40:41 +02:00
Ole André Vadla Ravnås f27a8dcc73 subprojects: Bump outdated 2026-08-16 22:23:39 +02:00
Ole André Vadla Ravnås 7c81781550 submodules: Bump releng 2026-08-16 22:23:39 +02:00
Ole André Vadla Ravnås 317a796f7d subprojects: Bump outdated 2026-08-13 18:00:26 +02:00
Ole André Vadla Ravnås 01f691c075 subprojects: Bump outdated 2026-08-13 17:55:09 +02:00
Ole André Vadla Ravnås feb036961e submodules: Bump releng 2026-08-13 17:55:09 +02:00
Ole André Vadla Ravnås 62f066389e subprojects: Bump outdated 2026-08-13 16:21:25 +02:00
Ole André Vadla Ravnås 8deda449c3 submodules: Bump releng 2026-08-13 16:21:25 +02:00
Ole André Vadla Ravnås 5aa3fda547 subprojects: Bump outdated 2026-08-13 13:35:02 +02:00
Ole André Vadla Ravnås d7acc26d9d submodules: Bump releng 2026-08-13 13:35:02 +02:00
Ole André Vadla Ravnås 8cb6565adf subprojects: Bump outdated 2026-08-13 13:02:05 +02:00
Ole André Vadla Ravnås 5ba902508e submodules: Bump releng 2026-08-13 13:02:05 +02:00
Ole André Vadla Ravnås 4ee43bc18c subprojects: Bump outdated 2026-08-13 00:19:40 +02:00
Ole André Vadla Ravnås 4414c049f4 submodules: Bump releng 2026-08-13 00:19:40 +02:00
Ole André Vadla Ravnås 1649a5a4cf subprojects: Bump outdated 2026-08-12 13:00:22 +02:00
Ole André Vadla Ravnås b17f7704a3 submodules: Bump releng 2026-08-12 13:00:22 +02:00
Ole André Vadla Ravnås 4aca013c22 subprojects: Bump outdated 2026-08-11 11:41:53 +02:00
Ole André Vadla Ravnås 1cc0bcae64 subprojects: Prepare for release 2026-08-05 00:32:55 +02:00
Ole André Vadla Ravnås 14849bcb1e subprojects: Bump outdated 2026-08-05 00:32:13 +02:00
Ole André Vadla Ravnås 20def06046 submodules: Bump releng 2026-08-05 00:32:13 +02:00
Ole André Vadla Ravnås 3ffd66c91d subprojects: Bump outdated 2026-08-02 18:29:21 +02:00
Ole André Vadla Ravnås da4e16c726 submodules: Bump releng 2026-08-02 18:29:21 +02:00
Ole André Vadla Ravnås 3353755177 Fix attach() linker_notifier_offsets kwarg
Its customization hardcoded an empty selector map instead of the one
computed from the options type, so the plural kwarg was rejected. Let
codegen build the options argument for custom facade methods that take
**kwargs.
2026-08-02 18:19:14 +02:00
Ole André Vadla Ravnås 82b4f9559f Fix Script.enable_debugger() port default
Vala's default argument is absent from the .gir, so the generated facade
made the port mandatory. Declare the default in the customizations, and
let param_typings override the facade signature also when there is no
custom logic.
2026-08-02 18:13:33 +02:00
Ole André Vadla Ravnås db4590d85c Fix certificate option on remote device et al
The GTlsCertificate-typed properties had no marshaller, so their setters
were never generated. Marshal them to and from a string holding either a
PEM blob or a filename, like the old hand-written bindings did.

Reported and diagnosed by rev1si0n <[email protected]>.
2026-07-31 00:31:04 +02:00
Ole André Vadla Ravnås 2a3a04b9be subprojects: Prepare for release 2026-07-22 00:30:14 +02:00
Ole André Vadla Ravnås 89e8222424 subprojects: Bump outdated 2026-07-22 00:29:03 +02:00
Ole André Vadla Ravnås e35a085928 bindgen: Restore remaining public names
Reinstate the cancellable decorator, RPCResult, make_rpc_call_request
and make_auth_callback, and expose ScriptExportsAsync from frida.core
via a new facade_epilogues customization.
2026-07-22 00:23:55 +02:00
Ole André Vadla Ravnås d1716f4a6c bindgen: Generate signal callback overloads
Facade on/off took any callable, so misshapen handlers type-checked.

Derive the stacks from the .gir: the alias name follows from the type
and signal name. Undefined aliases and single-signal types are skipped.
2026-07-22 00:23:55 +02:00
Ole André Vadla Ravnås 0646972686 bindgen: Restore missing facade members
The .gir yields only get_device_by_id/by_type, and drops
Cancellable.connect/disconnect, whose callback is not marshallable.

Reinstate all six. connect/disconnect go over the cancelled signal,
keeping g_cancellable_connect's return-0-when-cancelled contract.
2026-07-22 00:23:55 +02:00
Ole André Vadla Ravnås b51406984d bindgen: Restore facade typing surface
The .gir rewrite emitted only GObject-derived classes and no
annotations, so py.typed shipped an untyped facade.

Reinstate the aliases as a prelude asset and annotate the facade
from the existing pyi type mapping.

Also fixes read-only options properties and the module functions
missing from _frida.pyi.
2026-07-22 00:23:55 +02:00
Ole André Vadla Ravnås 082c9dfbc9 subprojects: Prepare for release 2026-07-20 17:56:01 +02:00
Ole André Vadla Ravnås 274c52291b bindgen: Fix formatting 2026-07-20 17:54:25 +02:00
Ole André Vadla Ravnås b9e24033ed subprojects: Bump outdated 2026-07-20 17:47:03 +02:00
Ole André Vadla Ravnås 10aeb056c6 bindgen: Fix options kwargs marshalling
Skip None values instead of passing them to typed setters.

Map collection kwargs onto their select_/add_ methods, walking
the parent chain. Affects specs, omits, externals, pids and
identifiers.
2026-07-20 17:45:06 +02:00
Ole André Vadla Ravnås e5e0558250 subprojects: Prepare for release 2026-07-19 22:25:25 +02:00
Ole André Vadla Ravnås cd74f3618a subprojects: Bump outdated 2026-07-19 22:24:35 +02:00
Ole André Vadla Ravnås 9fff6507af subprojects: Prepare for release 2026-07-18 13:30:31 +02:00
Ole André Vadla Ravnås 37badc7959 subprojects: Bump outdated 2026-07-18 13:29:18 +02:00
Ole André Vadla Ravnås 8b05045785 submodules: Bump releng 2026-07-18 13:29:18 +02:00
38 changed files with 1094 additions and 295 deletions

No files matched your search

+4 -2
View File
@@ -42,7 +42,8 @@ class Application:
print("✔ enable_child_gating()")
session.enable_child_gating()
print("✔ create_script()")
script = session.create_script("""\
script = session.create_script(
"""\
Interceptor.attach(Module.getGlobalExportByName('open'), {
onEnter(args) {
send({
@@ -51,7 +52,8 @@ Interceptor.attach(Module.getGlobalExportByName('open'), {
});
}
});
""")
"""
)
script.on("message", lambda message, data: self._reactor.schedule(lambda: self._on_message(pid, message)))
print("✔ load()")
script.load()
+4 -2
View File
@@ -1,7 +1,8 @@
import frida
session = frida.attach("Twitter")
script = session.create_script("""\
script = session.create_script(
"""\
rpc.exports = {
hello: function () {
return 'Hello';
@@ -10,7 +11,8 @@ rpc.exports = {
oops;
}
};
""")
"""
)
script.load()
api = script.exports_sync
print("api.hello() =>", api.hello())
+4 -2
View File
@@ -22,7 +22,8 @@ class Application:
self._session = session
session.on("detached", lambda *args: self._reactor.schedule(lambda: self._on_detached(*args)))
script = session.create_script("""
script = session.create_script(
"""
let _puts = null;
Interceptor.attach(DebugSymbol.getFunctionByName('f'), {
@@ -47,7 +48,8 @@ function puts(s) {
}
_puts(Memory.allocUtf8String(s));
}
""")
"""
)
self._script = script
script.on("message", lambda *args: self._reactor.schedule(lambda: self._on_message(*args)))
script.load()
+4 -2
View File
@@ -29,7 +29,8 @@ class Application:
session = self._device.attach(pid)
session.on("detached", lambda reason: self._reactor.schedule(lambda: self._on_detached(pid, session, reason)))
print("✔ create_script()")
script = session.create_script("""\
script = session.create_script(
"""\
const puts = new NativeFunction(Module.getGlobalExportByName('puts'), 'int', ['pointer']);
puts(Memory.allocUtf8String('Hello from Frida agent'));
Interceptor.attach(Module.getGlobalExportByName('open'), {
@@ -40,7 +41,8 @@ Interceptor.attach(Module.getGlobalExportByName('open'), {
});
}
});
""")
"""
)
script.on("message", lambda message, data: self._reactor.schedule(lambda: self._on_message(pid, message)))
print("✔ load()")
script.load()
@@ -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);
+5 -5
View File
@@ -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:
+6 -6
View File
@@ -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))
+7 -7
View File
@@ -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)
+31 -15
View File
@@ -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
+106
View File
@@ -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"]
+426 -59
View File
@@ -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,8 @@ def generate_extension_c(model: Model) -> str:
code += generate_interface_implementation(otype)
code += generate_object_type_constructor(otype)
for ctor in factory_constructors(otype):
code += generate_factory_function(otype, ctor, build_constructor_marshals(ctor))
return code
@@ -1050,6 +1239,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);",
@@ -1110,13 +1303,15 @@ def generate_object_type_toplevel_definitions(model: Model) -> str:
cprefix = otype.c_symbol_prefix
if otype.name == "Object":
defs.append(f"""PYFRIDA_DEFINE_BASETYPE ("_frida.{otype.py_name}", {otype.py_name}, g_object_unref,
defs.append(
f"""PYFRIDA_DEFINE_BASETYPE ("_frida.{otype.py_name}", {otype.py_name}, g_object_unref,
{{ Py_tp_doc, "{otype.name}" }},
{{ Py_tp_init, {cprefix}_init }},
{{ Py_tp_dealloc, {cprefix}_dealloc }},
{{ Py_tp_methods, {cprefix}_methods }},
{{ Py_tp_getset, {cprefix}_getsets }},
);""")
);"""
)
continue
parent = otype.parent
@@ -1136,12 +1331,14 @@ def generate_object_type_toplevel_definitions(model: Model) -> str:
)
continue
defs.append(f"""PYFRIDA_DEFINE_TYPE ("_frida.{otype.py_name}", {otype.py_name}, {parent_name}, g_object_unref,
defs.append(
f"""PYFRIDA_DEFINE_TYPE ("_frida.{otype.py_name}", {otype.py_name}, {parent_name}, g_object_unref,
{{ Py_tp_doc, "{otype.name}" }},
{{ Py_tp_init, {cprefix}_init }},
{{ Py_tp_methods, {cprefix}_methods }},
{{ Py_tp_getset, {cprefix}_getsets }},
);""")
);"""
)
return "\n\n".join(defs)
@@ -1166,6 +1363,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 +1381,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 +1729,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 +1758,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 +1774,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 +1827,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 +1925,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 +2113,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 +2203,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 +2351,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 +2500,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 +2595,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}
"""
@@ -2403,10 +2768,12 @@ def generate_object_type_structs(model: Model) -> str:
for otype in model.regular_object_types:
if otype.name == "Object":
continue
structs.append(f"""struct _{otype.c_symbol_prefix}
structs.append(
f"""struct _{otype.c_symbol_prefix}
{{
{otype.parent_c_symbol_prefix} parent;
}};""")
}};"""
)
return "\n\n".join(structs)
+58 -42
View File
@@ -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",),
)
+18 -4
View File
@@ -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: f3c2f0efbf...4e51c61880.
+1 -1
View File
@@ -1,6 +1,6 @@
[wrap-git]
url = https://github.com/frida/frida-core.git
revision = 17.16.0
revision = 17.23.2
depth = 1
[provide]
+54
View File
@@ -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