mirror of
https://github.com/frida/frida-python.git
synced 2026-11-11 10:01:27 -10:00
Compare commits
8
Commits
debug/tasks
...
17.17.0
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1cc0bcae64 | ||
|
|
14849bcb1e | ||
|
|
20def06046 | ||
|
|
3ffd66c91d | ||
|
|
da4e16c726 | ||
|
|
3353755177 | ||
|
|
82b4f9559f | ||
|
|
db4590d85c |
No files matched your search
@@ -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;
|
||||
|
||||
@@ -1041,6 +1041,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);
|
||||
|
||||
@@ -426,11 +426,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)
|
||||
|
||||
@@ -446,6 +447,20 @@ def generate_custom_facade_method(
|
||||
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")
|
||||
@@ -627,9 +642,7 @@ 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(facade_param(param, model))
|
||||
params = [facade_param(param, model) for param in method.input_parameters]
|
||||
names = ", ".join(param.name for param in method.input_parameters)
|
||||
|
||||
call = f"self._impl.{method.name}({names})"
|
||||
@@ -638,10 +651,14 @@ def generate_py_sync_method(method: Method, model: Model) -> Optional[str]:
|
||||
|
||||
ret = "None" if method.return_value is None else pyi_type(method.return_value.type, model)
|
||||
|
||||
return f""" def {method.name}({", ".join(signature)}) -> {ret}:
|
||||
return 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_param(param, model: Model) -> str:
|
||||
annotation = pyi_type(param.type, model)
|
||||
if param.nullable:
|
||||
@@ -669,7 +686,7 @@ def generate_py_async_method(method: Method, model: Model) -> Optional[str]:
|
||||
|
||||
|
||||
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":
|
||||
@@ -678,20 +695,24 @@ 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, {build_option_selectors(options)})")
|
||||
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}: Optional[{pyi_type(param.type, model)}] = None")
|
||||
params.append(f"{param.name}: Optional[{pyi_type(param.type, model)}] = None")
|
||||
args.append(f"_unwrap({param.name})")
|
||||
else:
|
||||
signature.append(facade_param(param, model))
|
||||
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:
|
||||
@@ -860,6 +881,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)
|
||||
@@ -1994,6 +2017,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"
|
||||
@@ -2073,6 +2107,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:
|
||||
|
||||
@@ -119,11 +119,7 @@ if aux:
|
||||
),
|
||||
"attach": MethodCustomizations(
|
||||
param_typings=["target: ProcessTarget", "**kwargs: Any"],
|
||||
custom_logic=(
|
||||
"pid = self._pid_of(target)\n" "options = _make_options(_frida.SessionOptions, kwargs, {})",
|
||||
"pid = await self._pid_of(target)\n"
|
||||
"options = _make_options(_frida.SessionOptions, kwargs, {})",
|
||||
),
|
||||
custom_logic=_RESOLVE_PID,
|
||||
),
|
||||
},
|
||||
custom_code=CustomCode(
|
||||
@@ -141,6 +137,7 @@ 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),
|
||||
},
|
||||
),
|
||||
|
||||
@@ -144,11 +144,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
|
||||
|
||||
|
||||
+1
-1
Submodule releng updated: abb8238970...c9cc18138b.
@@ -1,6 +1,6 @@
|
||||
[wrap-git]
|
||||
url = https://github.com/frida/frida-core.git
|
||||
revision = 17.16.4
|
||||
revision = 17.17.0
|
||||
depth = 1
|
||||
|
||||
[provide]
|
||||
|
||||
@@ -17,6 +17,27 @@ BINDGEN_CORE = REPO / "frida-bindgen"
|
||||
GIRDIR = REPO / "build" / "subprojects" / "frida-core" / "src" / "api"
|
||||
OUTDIR = REPO / "build" / "bindgen-out"
|
||||
BUILT_EXTENSION = REPO / "build" / "frida" / "_frida.abi3.so"
|
||||
CERTIFICATE = """\
|
||||
-----BEGIN CERTIFICATE-----
|
||||
MIIDGzCCAgOgAwIBAgIUEoCZX+FRWNTzcvvrCE2chFvE0EgwDQYJKoZIhvcNAQEL
|
||||
BQAwHDEaMBgGA1UEAwwRZnJpZGEtcHl0aG9uLXRlc3QwIBcNMjYwNzMwMjIyNjAy
|
||||
WhgPMjEyNjA3MDYyMjI2MDJaMBwxGjAYBgNVBAMMEWZyaWRhLXB5dGhvbi10ZXN0
|
||||
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAtKTIUbHLhkqRKdKqXuMc
|
||||
u7q8OZD2eKdykp4Qvv+I5JcmNW7nlW+58zTCLaqUCsSrojDb8T2qvtRJbaJYAofE
|
||||
P7bhb5ROOVwsyOjhvP7rZ4zRXIDAUlB7zE15yNkWL+Rh4l3wRxMOMIB2HQlurtDk
|
||||
PIqPOITwRvrADKZt/73iMASwrkkqKnl/F3rLT0JQrd907sBdGGZNbJM9L4TUBT9t
|
||||
MBBl6qXtDEFyBfeoqlnrLUTN/Jwyka+/6n6eZCN8g2g5tS1Lc7LFjX4iBG9ibgaH
|
||||
hrNSTM+haLImFS3dtXFr4rOCAHnDYYvkCY1nuZOXannCvZowv0xczqpGLbj5IYJK
|
||||
MQIDAQABo1MwUTAdBgNVHQ4EFgQUi83cV5yMTdWaVeRpgNO3vyeitMMwHwYDVR0j
|
||||
BBgwFoAUi83cV5yMTdWaVeRpgNO3vyeitMMwDwYDVR0TAQH/BAUwAwEB/zANBgkq
|
||||
hkiG9w0BAQsFAAOCAQEADScSFuasB4mgOb/kktq3O09rf/uUIa6OEjdFRec/i2+E
|
||||
ZAUd5onsWQTHm4yhjqQwqVBte58tTqZB7S1HMCy5sEgEVUFnTrMUqtrf/WAnsWA6
|
||||
SdjNr7h/aRnLHHysLkbl0Zo4fOIWIYSJpLn39e0OzcNBjEHN5U9HE/m/2LFku1Kb
|
||||
KMTYXrhBKvVPpZLxZyoLz00EM7euaNx3juAgfQdSDfKGahf5CqqMJ/VmITISDJDR
|
||||
jtN3wcv0/nZA5x/b73QCj0jqIEWqguYf4GLWitxGsSdT2dBEpT88Kz3Ibwb8uRT+
|
||||
Ls3eRsdoqxiOT0E7Z44O8gjbqsYOr/jdMShwEytDwQ==
|
||||
-----END CERTIFICATE-----
|
||||
"""
|
||||
|
||||
|
||||
@unittest.skipUnless(
|
||||
@@ -131,6 +152,29 @@ class TestExtension(unittest.TestCase):
|
||||
options.argv = None
|
||||
self.assertIsNone(options.argv)
|
||||
|
||||
def test_certificate_property_accepts_pem_and_path(self):
|
||||
import tempfile
|
||||
|
||||
options = self._frida.RemoteDeviceOptions()
|
||||
self.assertIsNone(options.certificate)
|
||||
|
||||
options.certificate = CERTIFICATE
|
||||
self.assertEqual(options.certificate, CERTIFICATE)
|
||||
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".pem") as f:
|
||||
f.write(CERTIFICATE)
|
||||
f.flush()
|
||||
options.certificate = f.name
|
||||
self.assertEqual(options.certificate, CERTIFICATE)
|
||||
|
||||
options.certificate = None
|
||||
self.assertIsNone(options.certificate)
|
||||
|
||||
def test_certificate_property_rejects_missing_file(self):
|
||||
options = self._frida.RemoteDeviceOptions()
|
||||
with self.assertRaises(self._frida.InvalidArgumentError):
|
||||
options.certificate = "/nonexistent.pem"
|
||||
|
||||
def test_vardict_property_round_trips(self):
|
||||
options = self._frida.SpawnOptions()
|
||||
options.aux = {"name": "value", "count": 42, "flag": True}
|
||||
@@ -646,6 +690,16 @@ class TestFacade(unittest.TestCase):
|
||||
finally:
|
||||
session.detach()
|
||||
|
||||
def test_attach_accepts_linker_notifier_offsets(self):
|
||||
session = self.frida.get_local_device().attach(0, linker_notifier_offsets=[0x100])
|
||||
session.detach()
|
||||
|
||||
def test_enable_debugger_defaults_to_any_port(self):
|
||||
import inspect
|
||||
|
||||
for script in [self.frida.Script, self.frida.aio.Script]:
|
||||
self.assertEqual(inspect.signature(script.enable_debugger).parameters["port"].default, 0)
|
||||
|
||||
def test_file_monitor_reports_changes(self):
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
Reference in new issue
Block a user