Compare commits

..
1 Commits
Author SHA1 Message Date
Ole André Vadla Ravnås b7e22441ef Add print_alive_tasks() debug binding
Expose g_task_print_alive_tasks() as _frida.print_alive_tasks(),
wrapped as frida.print_alive_tasks() in both facades. Requires a
GLib built with G_ENABLE_DEBUG.
2026-07-28 16:42:54 +02:00
17 changed files with 133 additions and 449 deletions

No files matched your search

+2 -4
View File
@@ -42,8 +42,7 @@ 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({
@@ -52,8 +51,7 @@ Interceptor.attach(Module.getGlobalExportByName('open'), {
});
}
});
"""
)
""")
script.on("message", lambda message, data: self._reactor.schedule(lambda: self._on_message(pid, message)))
print("✔ load()")
script.load()
+2 -4
View File
@@ -1,8 +1,7 @@
import frida
session = frida.attach("Twitter")
script = session.create_script(
"""\
script = session.create_script("""\
rpc.exports = {
hello: function () {
return 'Hello';
@@ -11,8 +10,7 @@ rpc.exports = {
oops;
}
};
"""
)
""")
script.load()
api = script.exports_sync
print("api.hello() =>", api.hello())
+2 -4
View File
@@ -22,8 +22,7 @@ class Application:
self._session = session
session.on("detached", lambda *args: self._reactor.schedule(lambda: self._on_detached(*args)))
script = session.create_script(
"""
script = session.create_script("""
let _puts = null;
Interceptor.attach(DebugSymbol.getFunctionByName('f'), {
@@ -48,8 +47,7 @@ function puts(s) {
}
_puts(Memory.allocUtf8String(s));
}
"""
)
""")
self._script = script
script.on("message", lambda *args: self._reactor.schedule(lambda: self._on_message(*args)))
script.load()
+2 -4
View File
@@ -29,8 +29,7 @@ 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'), {
@@ -41,8 +40,7 @@ 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,3 +1,27 @@
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,
@@ -7,7 +31,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;
PyObject * certificate_value = NULL;
char * certificate_value = NULL;
char * origin = NULL;
PyObject * auth_service_obj = NULL;
char * asset_root_value = NULL;
@@ -19,10 +43,10 @@ PyEndpointParameters_init (PyEndpointParameters * self,
if (PyGObject_tp_init ((PyObject *) self, args, kw) < 0)
return -1;
if (!PyArg_ParseTupleAndKeywords (args, kw, "|esHOesOes", keywords,
if (!PyArg_ParseTupleAndKeywords (args, kw, "|esHesesOes", keywords,
"utf-8", &address,
&port,
&certificate_value,
"utf-8", &certificate_value,
"utf-8", &origin,
&auth_service_obj,
"utf-8", &asset_root_value))
@@ -50,6 +74,7 @@ beach:
PyMem_Free (asset_root_value);
PyMem_Free (origin);
PyMem_Free (certificate_value);
PyMem_Free (address);
return result;
@@ -910,9 +910,6 @@ 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:
@@ -1044,64 +1041,6 @@ 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,8 +35,6 @@ 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);
@@ -77,3 +77,8 @@ def make_auth_callback(callback: Callable[[str], Any]) -> Callable[[Any], str]:
return json.dumps(session_info)
return authenticate
# TEMPORARY DEBUG HACK
def print_alive_tasks() -> None:
_frida.print_alive_tasks()
@@ -78,3 +78,8 @@ def make_auth_callback(callback: Callable[[str], Any]) -> Callable[[Any], str]:
return json.dumps(session_info)
return authenticate
# TEMPORARY DEBUG HACK
def print_alive_tasks() -> None:
_frida.print_alive_tasks()
@@ -0,0 +1,11 @@
/* TEMPORARY DEBUG HACK: needs a GLib built with G_ENABLE_DEBUG. */
extern void g_task_print_alive_tasks (void);
static PyObject *
PyFrida_print_alive_tasks (PyObject * module,
PyObject * args)
{
g_task_print_alive_tasks ();
Py_RETURN_NONE;
}
+58 -300
View File
@@ -34,14 +34,10 @@ 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"""
FACADE_TYPING_IMPORTS = (
"from typing import (Any, Callable, Dict, List, Literal, Mapping, NotRequired, Optional, Tuple, TypedDict, "
"Sequence, Union, ParamSpec, TypeVar, cast, overload)"
)
def generate_facade_preludes(model: Model) -> List[str]:
@@ -92,25 +88,17 @@ def _to_envp(value):
def _make_options(cls, values, selectors):
options = cls()
_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))
setattr(options, name, value)
else:
add = getattr(impl, select)
add = getattr(options, select)
for element in value:
add(_unwrap(element))
add(element)
return options
_current_cancellable = threading.local()
@@ -194,25 +182,17 @@ def _to_envp(value):
def _make_options(cls, values, selectors):
options = cls()
_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))
setattr(options, name, value)
else:
add = getattr(impl, select)
add = getattr(options, select)
for element in value:
add(_unwrap(element))
add(element)
return options
_current_cancellable: contextvars.ContextVar = contextvars.ContextVar("frida_current_cancellable", default=None)
@@ -436,7 +416,7 @@ 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)
ret = facade_return_type(method, model)
ret = "None" if method.return_value is None else pyi_type(method.return_value.type, model)
return f""" async def {method.name}({signature}) -> {ret}:
return {call}"""
@@ -446,12 +426,11 @@ 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}, ({args}))"
invoke = f"_invoke(self._impl.{method.name}, ({method.facade_call_args}))"
call = f"await {invoke}" if awaitable else invoke
else:
call = f"self._impl.{method.name}({args})"
call = f"self._impl.{method.name}({method.facade_call_args})"
if method.return_value is not None:
call = wrap_result(call, method.return_value.type, model)
@@ -460,27 +439,13 @@ def generate_custom_facade_method(
body = textwrap.indent(logic.strip(), " " * 8)
keyword = "async def" if awaitable else "def"
ret = facade_return_type(method, model)
ret = "None" if method.return_value is None else pyi_type(method.return_value.type, model)
return f""" {keyword} {method.name}({signature}) -> {ret}:
{body}
return {call}"""
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")
@@ -536,8 +501,10 @@ 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])
_apply_settings(self._impl, kwargs, {build_option_selectors(otype)})
self._impl = _frida.{otype.py_name}(
*[_unwrap(a) for a in args],
**{{k: _unwrap(v) for k, v in kwargs.items()}},
)
setup = getattr(self, "_setup", None)
if setup is not None:
setup()"""
@@ -660,29 +627,21 @@ 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
params = [facade_param(param, model) for param in method.input_parameters]
names = ", ".join(facade_argument(param, model) for param in method.input_parameters)
signature = ["self"]
for param in method.input_parameters:
signature.append(facade_param(param, model))
names = ", ".join(param.name for param in method.input_parameters)
call = f"self._impl.{method.name}({names})"
if method.return_value is not None:
call = wrap_result(call, method.return_value.type, model)
ret = facade_return_type(method, model)
ret = "None" if method.return_value is None else pyi_type(method.return_value.type, model)
return f""" def {method.name}({facade_signature(method, params)}) -> {ret}:
return f""" def {method.name}({", ".join(signature)}) -> {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:
@@ -703,14 +662,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)
ret = facade_return_type(method, model)
ret = "None" if method.return_value is None else pyi_type(method.return_value.type, model)
return f""" def {method.name}({signature}) -> {ret}:
return {call}"""
def build_facade_async_parts(method: Method, model: Model) -> Optional[Tuple[str, str]]:
params = []
signature = ["self"]
args = []
for param in method.input_parameters:
if param.type.name == "Gio.Cancellable":
@@ -719,24 +678,20 @@ 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:
params.append("**kwargs")
args.append(make_options_expr(options))
signature.append("**kwargs")
args.append(f"_make_options(_frida.{options.py_name}, kwargs, {build_option_selectors(options)})")
elif resolve_input_object_type(param.type, model) is not None:
params.append(f"{param.name}: Optional[{pyi_type(param.type, model)}] = None")
signature.append(f"{param.name}: Optional[{pyi_type(param.type, model)}] = None")
args.append(f"_unwrap({param.name})")
else:
params.append(facade_param(param, model))
signature.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 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)})"
return ", ".join(signature), "".join(arg + ", " for arg in args)
def build_option_selectors(options: ObjectType) -> str:
@@ -793,22 +748,13 @@ def generate_pyi_class(otype: ObjectType, model: Model) -> List[str]:
body = []
primary = inline_constructor(otype)
if primary is not None:
params = pyi_parameters(primary.input_parameters, model)
ctor = next(iter(otype.constructors), None)
if ctor is not None and not ctor.throws:
params = pyi_parameters(ctor.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: ...")
@@ -869,13 +815,7 @@ def pyi_method(method: Method, model: Model) -> Optional[str]:
if params is None:
return None
signature = ", ".join(["self"] + params)
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)
ret = "None" if method.return_value is None else pyi_type(method.return_value.type, model)
return f" def {method.name}({signature}) -> {ret}: ..."
@@ -920,8 +860,6 @@ 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)
@@ -963,8 +901,6 @@ 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
@@ -1239,10 +1175,6 @@ 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);",
@@ -1303,15 +1235,13 @@ 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
@@ -1331,14 +1261,12 @@ 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)
@@ -1363,9 +1291,6 @@ 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[] =
{{
@@ -1381,11 +1306,6 @@ 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)
@@ -1729,17 +1649,14 @@ 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 and out is None:
if not params and not method.throws and not returns_strv:
if method.return_value is None:
body = f" {call};\n\n PyFrida_RETURN_NONE;"
else:
@@ -1758,8 +1675,6 @@ 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()]
@@ -1774,15 +1689,7 @@ static PyObject *
pre = "".join(f"\n {p.pre}\n" for p in params if p.pre)
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:
if method.return_value is None:
invocation = f" {call};"
success = " Py_IncRef (Py_None);\n result = Py_None;"
elif returns_strv:
@@ -1827,22 +1734,6 @@ 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
@@ -1925,16 +1816,6 @@ 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
@@ -2113,17 +1994,6 @@ 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"
@@ -2203,8 +2073,6 @@ 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:
@@ -2351,55 +2219,28 @@ def generate_object_type_constructor(otype: ObjectType) -> str:
if implementable_interface(otype):
return generate_interface_init(otype)
ctor = primary_constructor(otype)
ctor = next(iter(otype.constructors), None)
if ctor is None:
return generate_bare_init(otype)
if ctor.throws:
return generate_unconstructable_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)
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)
@@ -2500,88 +2341,7 @@ static int
{generate_constructor_tail(marshals)}"""
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:
def generate_argument_parse(marshals: List["ParamMarshal"]) -> str:
fmt = ""
optional_started = False
for m in marshals:
@@ -2595,7 +2355,7 @@ def generate_argument_parse(marshals: List["ParamMarshal"], on_failure: str = "r
indent = " " * len(" if (!PyArg_ParseTupleAndKeywords (")
parse_args_str = (",\n" + indent).join(parse_args)
return f""" if (!PyArg_ParseTupleAndKeywords ({parse_args_str}))
{on_failure}
return -1;
"""
@@ -2768,12 +2528,10 @@ 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)
+11 -2
View File
@@ -34,6 +34,12 @@ def load_customizations() -> Customizations:
c_symbol="PyFrida_get_device_manager",
asset="module_get_device_manager.c",
),
# TEMPORARY DEBUG HACK
ModuleFunction(
py_name="print_alive_tasks",
c_symbol="PyFrida_print_alive_tasks",
asset="module_print_alive_tasks.c",
),
),
),
),
@@ -119,7 +125,11 @@ if aux:
),
"attach": MethodCustomizations(
param_typings=["target: ProcessTarget", "**kwargs: Any"],
custom_logic=_RESOLVE_PID,
custom_logic=(
"pid = self._pid_of(target)\n" "options = _make_options(_frida.SessionOptions, kwargs, {})",
"pid = await self._pid_of(target)\n"
"options = _make_options(_frida.SessionOptions, kwargs, {})",
),
),
},
custom_code=CustomCode(
@@ -137,7 +147,6 @@ if aux:
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),
},
),
+4 -8
View File
@@ -17,10 +17,6 @@ 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
@@ -148,6 +144,10 @@ 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
@@ -281,7 +281,6 @@ def _make_class(
type_struct,
parent,
constructors,
functions,
methods,
properties,
signals,
@@ -296,7 +295,6 @@ def _make_class(
type_struct,
parent,
constructors,
functions,
methods,
properties,
signals,
@@ -313,7 +311,6 @@ def _make_interface(
type_struct,
parent,
constructors,
functions,
methods,
properties,
signals,
@@ -327,7 +324,6 @@ def _make_interface(
type_struct,
parent,
constructors,
functions,
methods,
properties,
signals,
+1 -1
Submodule releng updated: 4e51c61880...abb8238970.
+1 -1
View File
@@ -1,6 +1,6 @@
[wrap-git]
url = https://github.com/frida/frida-core.git
revision = 17.23.3
revision = 17.16.4
depth = 1
[provide]
-54
View File
@@ -17,27 +17,6 @@ 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(
@@ -152,29 +131,6 @@ 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}
@@ -690,16 +646,6 @@ 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