mirror of
https://github.com/frida/frida-python.git
synced 2026-11-07 09:35:57 -10:00
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.
This commit is contained in:
1 parent
d1716f4a6c
commit
e35a085928
10 files changed
+95
-1
No files matched your search
@@ -21,3 +21,20 @@ class CancellablePollFD:
|
||||
|
||||
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
|
||||
@@ -21,3 +21,20 @@ class CancellablePollFD:
|
||||
|
||||
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
|
||||
@@ -0,0 +1 @@
|
||||
ScriptExportsAsync = aio._ScriptExports
|
||||
@@ -40,3 +40,20 @@ class _ScriptExports:
|
||||
|
||||
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)
|
||||
@@ -36,3 +36,20 @@ class _ScriptExports:
|
||||
|
||||
|
||||
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)
|
||||
@@ -69,3 +69,11 @@ def inject_library_blob(target: ProcessTarget, blob: bytes, entrypoint: str, dat
|
||||
|
||||
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
|
||||
@@ -70,3 +70,11 @@ async def inject_library_blob(target: ProcessTarget, blob: bytes, entrypoint: st
|
||||
|
||||
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
|
||||
@@ -36,7 +36,7 @@ def read_asset(name: str) -> str:
|
||||
|
||||
FACADE_TYPING_IMPORTS = (
|
||||
"from typing import (Any, Callable, Dict, List, Literal, Mapping, NotRequired, Optional, Tuple, TypedDict, "
|
||||
"Union, overload)"
|
||||
"Sequence, Union, ParamSpec, TypeVar, cast, overload)"
|
||||
)
|
||||
|
||||
|
||||
@@ -244,7 +244,9 @@ 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",
|
||||
@@ -280,6 +282,9 @@ def generate_py(model: Model) -> str:
|
||||
lines.append("")
|
||||
lines.append("")
|
||||
lines.append("from . import aio")
|
||||
for asset in model.customizations.facade_epilogues:
|
||||
lines.append("")
|
||||
lines.append(read_asset(asset).strip())
|
||||
lines.append("")
|
||||
lines.append("core = sys.modules[__name__]")
|
||||
lines.append('sys.modules[__name__ + ".core"] = core')
|
||||
@@ -312,7 +317,9 @@ def generate_aio(model: Model) -> str:
|
||||
"import asyncio",
|
||||
"import contextvars",
|
||||
"import time",
|
||||
"import dataclasses",
|
||||
"import fnmatch",
|
||||
"import functools",
|
||||
"import inspect",
|
||||
"import json",
|
||||
"import sys",
|
||||
|
||||
@@ -399,4 +399,5 @@ if request_handler is not None:
|
||||
return Customizations(
|
||||
type_customizations=type_customizations,
|
||||
facade_preludes=("facade_types.py",),
|
||||
facade_epilogues=("facade_epilogue.py",),
|
||||
)
|
||||
@@ -201,6 +201,7 @@ class EnumerationMember(core.EnumerationMember):
|
||||
class Customizations:
|
||||
type_customizations: Mapping[str, TypeCustomizations] = field(default_factory=OrderedDict)
|
||||
facade_preludes: Tuple[str, ...] = ()
|
||||
facade_epilogues: Tuple[str, ...] = ()
|
||||
|
||||
|
||||
@dataclass
|
||||
|
||||
Reference in new issue
Block a user