虚拟环境

虚拟环境
This commit is contained in:
2023-08-12 16:46:30 +08:00
parent e20472de75
commit 3eef1ecdb7
2540 changed files with 452477 additions and 0 deletions

View File

@@ -0,0 +1,198 @@
import abc
import codecs
import json
import os
from typing import TYPE_CHECKING, Optional, Sequence
if TYPE_CHECKING:
import frida_tools.repl
class Magic(abc.ABC):
@property
def description(self) -> str:
return "no description"
@abc.abstractproperty
def required_args_count(self) -> int:
pass
def execute(self, repl: "frida_tools.repl.REPLApplication", args: Sequence[str]) -> Optional[bool]:
pass
class Resume(Magic):
@property
def description(self) -> str:
return "resume execution of the spawned process"
@property
def required_args_count(self) -> int:
return 0
def execute(self, repl: "frida_tools.repl.REPLApplication", args: Sequence[str]) -> None:
repl._reactor.schedule(lambda: repl._resume())
class Load(Magic):
@property
def description(self) -> str:
return "Load an additional script and reload the current REPL state"
@property
def required_args_count(self) -> int:
return 1
def execute(self, repl: "frida_tools.repl.REPLApplication", args: Sequence[str]) -> None:
try:
proceed = repl._get_confirmation(
"Are you sure you want to load a new script and discard all current state?"
)
if not proceed:
repl._print("Discarding load command")
return
repl._user_scripts.append(args[0])
repl._perform_on_reactor_thread(lambda: repl._load_script())
except Exception as e:
repl._print(f"Failed to load script: {e}")
class Reload(Magic):
@property
def description(self) -> str:
return "reload (i.e. rerun) the script that was given as an argument to the REPL"
@property
def required_args_count(self) -> int:
return 0
def execute(self, repl: "frida_tools.repl.REPLApplication", args: Sequence[str]) -> bool:
try:
repl._perform_on_reactor_thread(lambda: repl._load_script())
return True
except Exception as e:
repl._print(f"Failed to load script: {e}")
return False
class Unload(Magic):
@property
def required_args_count(self) -> int:
return 0
def execute(self, repl: "frida_tools.repl.REPLApplication", args: Sequence[str]) -> None:
repl._unload_script()
class Autoperform(Magic):
@property
def description(self) -> str:
return (
"receive on/off as first and only argument, when switched on will wrap any REPL code with Java.performNow()"
)
@property
def required_args_count(self) -> int:
return 1
def execute(self, repl: "frida_tools.repl.REPLApplication", args: Sequence[str]) -> None:
repl._autoperform_command(args[0])
class Autoreload(Magic):
_VALID_ARGUMENTS = ("on", "off")
@property
def description(self) -> str:
return "disable or enable auto reloading of script files"
@property
def required_args_count(self) -> int:
return 1
def execute(self, repl: "frida_tools.repl.REPLApplication", args: Sequence[str]) -> None:
if args[0] not in self._VALID_ARGUMENTS:
raise ValueError("Autoreload command only receive on or off as an argument")
required_state = args[0] == "on"
if required_state == repl._autoreload:
repl._print("Autoreloading is already in the desired state")
return
if required_state:
repl._monitor_all()
else:
repl._demonitor_all()
repl._autoreload = required_state
class Exec(Magic):
@property
def description(self) -> str:
return "execute the given file path in the context of the currently loaded scripts"
@property
def required_args_count(self) -> int:
return 1
def execute(self, repl: "frida_tools.repl.REPLApplication", args: Sequence[str]) -> None:
if not os.path.exists(args[0]):
repl._print("Can't read the given file because it does not exist")
return
try:
with codecs.open(args[0], "rb", "utf-8") as f:
if not repl._exec_and_print(repl._evaluate_expression, f.read()):
repl._errors += 1
except PermissionError:
repl._print("Can't read the given file because of a permission error")
class Time(Magic):
@property
def description(self) -> str:
return "measure the execution time of the given expression and print it to the screen"
@property
def required_args_count(self) -> int:
return -2
def execute(self, repl: "frida_tools.repl.REPLApplication", args: Sequence[str]) -> None:
repl._exec_and_print(
repl._evaluate_expression,
"""
(() => {{
const _startTime = Date.now();
const _result = eval({expression});
const _endTime = Date.now();
console.log('Time: ' + (_endTime - _startTime) + ' ms.');
return _result;
}})();""".format(
expression=json.dumps(" ".join(args))
),
)
class Help(Magic):
@property
def description(self) -> str:
return "print a list of available REPL commands"
@property
def required_args_count(self) -> int:
return 0
def execute(self, repl: "frida_tools.repl.REPLApplication", args: Sequence[str]) -> None:
repl._print("Available commands: ")
for name, command in repl._magic_command_args.items():
if command.required_args_count >= 0:
required_args = f"({command.required_args_count})"
else:
required_args = f"({abs(command.required_args_count) - 1}+)"
repl._print(f" %{name}{required_args} - {command.description}")
repl._print("")
repl._print("For help with Frida scripting API, check out https://frida.re/docs/")
repl._print("")

View File

@@ -0,0 +1,353 @@
from __future__ import annotations
import argparse
import os
import struct
from enum import IntEnum
from io import BufferedReader
from typing import List
from zipfile import ZipFile
def main() -> None:
from frida_tools.application import ConsoleApplication
class ApkApplication(ConsoleApplication):
def _usage(self) -> str:
return "%(prog)s [options] path.apk"
def _add_options(self, parser: argparse.ArgumentParser) -> None:
parser.add_argument("-o", "--output", help="output path", metavar="OUTPUT")
parser.add_argument("apk", help="apk file")
def _needs_device(self) -> bool:
return False
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
self._output_path = options.output
self._path = options.apk
if not self._path.endswith(".apk"):
parser.error("path must end in .apk")
if self._output_path is None:
self._output_path = self._path.replace(".apk", ".d.apk")
def _start(self) -> None:
try:
debug(self._path, self._output_path)
except Exception as e:
self._update_status(f"Error: {e}")
self._exit(1)
self._exit(0)
app = ApkApplication()
app.run()
def debug(path: str, output_path: str) -> None:
with ZipFile(path, "r") as iz, ZipFile(output_path, "w") as oz:
for info in iz.infolist():
with iz.open(info) as f:
if info.filename == "AndroidManifest.xml":
manifest = BinaryXML(f)
pool = None
debuggable_index = None
size = 8
for header in manifest.chunk_headers[1:]:
if header.type == ChunkType.STRING_POOL:
pool = StringPool(header)
debuggable_index = pool.append_str("debuggable")
if header.type == ChunkType.RESOURCE_MAP:
# The "debuggable" attribute name is not only a reference to the string pool, but
# also to the resource map. We need to extend the resource map with a valid entry.
# refs https://justanapplication.wordpress.com/category/android/android-binary-xml/android-xml-startelement-chunk/
resource_map = ResourceMap(header)
resource_map.add_debuggable(debuggable_index)
if header.type == ChunkType.START_ELEMENT:
start = StartElement(header)
name = pool.get_string(start.name)
if name == "application":
start.insert_debuggable(debuggable_index, resource_map)
size += header.size
header = manifest.chunk_headers[0]
header_data = bytearray(header.chunk_data)
header_data[4 : 4 + 4] = struct.pack("<I", size)
data = bytearray()
data.extend(header_data)
for header in manifest.chunk_headers[1:]:
data.extend(header.chunk_data)
oz.writestr(info.filename, bytes(data), info.compress_type)
elif info.filename.upper() == "META-INF/MANIFEST.MF":
# Historically frida-apk deleted META-INF/ entirely, but that breaks some apps.
# It turns out that v1 signatures (META-INF/MANIFEST.MF) are not validated at all on
# modern Android versions, so we can keep them in for now.
# If this doesn't work for you, try to comment out the following line.
oz.writestr(info.filename, f.read(), info.compress_type)
else:
oz.writestr(info.filename, f.read(), info.compress_type)
class BinaryXML:
def __init__(self, stream: BufferedReader) -> None:
self.stream = stream
self.chunk_headers = []
self.parse()
def parse(self) -> None:
chunk_header = ChunkHeader(self.stream, False)
if chunk_header.type != ChunkType.XML:
raise BadHeader()
self.chunk_headers.append(chunk_header)
size = chunk_header.size
while self.stream.tell() < size:
chunk_header = ChunkHeader(self.stream)
self.chunk_headers.append(chunk_header)
class ChunkType(IntEnum):
STRING_POOL = 0x001
XML = 0x003
START_ELEMENT = 0x102
RESOURCE_MAP = 0x180
class ResourceType(IntEnum):
BOOL = 0x12
class StringType(IntEnum):
UTF8 = 1 << 8
class BadHeader(Exception):
pass
class ChunkHeader:
FORMAT = "<HHI"
def __init__(self, stream: BufferedReader, consume_data: bool = True) -> None:
self.stream = stream
data = self.stream.peek(struct.calcsize(self.FORMAT))
(self.type, self.header_size, self.size) = struct.unpack_from(self.FORMAT, data)
if consume_data:
self.chunk_data = self.stream.read(self.size)
else:
self.chunk_data = self.stream.read(struct.calcsize(self.FORMAT))
class StartElement:
FORMAT = "<HHIIIIIIHHHH"
ATTRIBUTE_FORMAT = "<IIiHBBi"
def __init__(self, header: ChunkHeader) -> None:
self.header = header
self.stream = self.header.stream
self.header_size = struct.calcsize(self.FORMAT)
data = struct.unpack_from(self.FORMAT, self.header.chunk_data)
if data[0] != ChunkType.START_ELEMENT:
raise BadHeader()
self.name = data[6]
self.attribute_count = data[8]
attributes_data = self.header.chunk_data[self.header_size :]
if len(attributes_data[-20:]) == 20:
previous_attribute = struct.unpack(self.ATTRIBUTE_FORMAT, attributes_data[-20:])
self.namespace = previous_attribute[0]
else:
# There are no other attributes in the application tag
self.namespace = -1
def insert_debuggable(self, name: int, resource_map: ResourceMap) -> None:
# TODO: Instead of using the previous attribute to determine the probable
# namespace for the debuggable tag we could scan the strings section
# for the AndroidManifest schema tag
if self.namespace == -1:
raise BadHeader()
chunk_data = bytearray(self.header.chunk_data)
resource_size = 8
resource_type = ResourceType.BOOL
# Denotes a True value in AXML, 0 is used for False
resource_data = -1
debuggable = struct.pack(
self.ATTRIBUTE_FORMAT, self.namespace, name, -1, resource_size, 0, resource_type, resource_data
)
# Some parts of Android expect this to be sorted by resource ID.
attr_offset = None
for insert_pos in range(self.attribute_count + 1):
attr_offset = 0x24 + 20 * insert_pos
idx = int.from_bytes(chunk_data[attr_offset + 4 : attr_offset + 8], "little")
if resource_map.get_resource(idx) > ResourceMap.DEBUGGING_RESOURCE:
break
chunk_data[attr_offset:attr_offset] = debuggable
self.header.size = len(chunk_data)
chunk_data[4 : 4 + 4] = struct.pack("<I", self.header.size)
self.attribute_count += 1
chunk_data[28 : 28 + 2] = struct.pack("<H", self.attribute_count)
self.header.chunk_data = bytes(chunk_data)
class ResourceMap:
DEBUGGING_RESOURCE = 0x101000F
def __init__(self, header: ChunkHeader) -> None:
self.header = header
def add_debuggable(self, idx: int) -> None:
assert idx is not None
data_size = len(self.header.chunk_data) - 8
target = (idx + 1) * 4
self.header.chunk_data += b"\x00" * (target - data_size - 4) + self.DEBUGGING_RESOURCE.to_bytes(4, "little")
self.header.size = len(self.header.chunk_data)
self.header.chunk_data = (
self.header.chunk_data[:4] + struct.pack("<I", self.header.size) + self.header.chunk_data[8:]
)
def get_resource(self, index: int) -> int:
offset = index * 4 + 8
return int.from_bytes(self.header.chunk_data[offset : offset + 4], "little")
class StringPool:
FORMAT = "<HHIIIIII"
def __init__(self, header: ChunkHeader):
self.header = header
self.stream = self.header.stream
self.header_size = struct.calcsize(self.FORMAT)
data = struct.unpack_from(self.FORMAT, self.header.chunk_data)
if data[0] != ChunkType.STRING_POOL:
raise BadHeader()
self.string_count = data[3]
self.flags = data[5]
self.strings_offset = data[6]
self.styles_offset = data[7]
self.utf8 = (self.flags & StringType.UTF8) != 0
self.dirty = False
offsets_data = self.header.chunk_data[self.header_size : self.header_size + self.string_count * 4]
self.offsets: List[int] = list(map(lambda f: f[0], struct.iter_unpack("<I", offsets_data)))
def get_string(self, index: int) -> str:
offset = self.offsets[index]
# HACK: We subtract 4 because we insert a string offset during append_str
# but we do not update the original stream and thus it reads stale data.
if self.dirty:
offset -= 4
position = self.stream.tell()
self.stream.seek(self.strings_offset + 8 + offset, os.SEEK_SET)
string = None
if self.utf8:
# Ignore number of characters
n = struct.unpack("<B", self.stream.read(1))[0]
if n & 0x80:
n = ((n & 0x7F) << 8) | struct.unpack("<B", self.stream.read(1))[0]
# UTF-8 encoded length
n = struct.unpack("<B", self.stream.read(1))[0]
if n & 0x80:
n = ((n & 0x7F) << 8) | struct.unpack("<B", self.stream.read(1))[0]
string = self.stream.read(n).decode("utf-8")
else:
n = struct.unpack("<H", self.stream.read(2))[0]
if n & 0x8000:
n |= ((n & 0x7FFF) << 16) | struct.unpack("<H", self.stream.read(2))[0]
string = self.stream.read(n * 2).decode("utf-16le")
self.stream.seek(position, os.SEEK_SET)
return string
def append_str(self, add: str) -> int:
data_size = len(self.header.chunk_data)
# Reserve data for our new offset
data_size += 4
chunk_data = bytearray(data_size)
end = self.header_size + self.string_count * 4
chunk_data[:end] = self.header.chunk_data[:end]
chunk_data[end + 4 :] = self.header.chunk_data[end:]
# Add 4 since we have added a string offset
offset = len(chunk_data) - 8 - self.strings_offset + 4
if self.utf8:
assert len(add.encode("utf-8")) < 128 # multi-byte len strings not supported yet
length_in_characters = len(add)
length_in_bytes = len(add.encode("utf-8"))
chunk_data.extend(struct.pack("<BB", length_in_characters, length_in_bytes))
chunk_data.extend(add.encode("utf-8"))
# Insert a UTF-8 NUL
chunk_data.extend([0])
else:
chunk_data.extend(struct.pack("<H", len(add)))
chunk_data.extend(add.encode("utf-16le"))
# Insert a UTF-16 NUL
chunk_data.extend([0, 0])
# pad to a multiple of 4 bytes
if len(chunk_data) % 4 != 0:
alignment_padding = [0] * (4 - len(chunk_data) % 4)
chunk_data.extend(alignment_padding)
# Insert a new offset at the end of the existing offsets
chunk_data[end : end + 4] = struct.pack("<I", offset)
# Increase the header size since we have inserted a new offset and string
self.header.size = len(chunk_data)
chunk_data[4 : 4 + 4] = struct.pack("<I", self.header.size)
self.string_count += 1
chunk_data[8 : 8 + 4] = struct.pack("<I", self.string_count)
# Increase strings offset since we have inserted a new offset and thus
# shifted the offset of the strings
self.strings_offset += 4
chunk_data[20 : 20 + 4] = struct.pack("<I", self.strings_offset)
# If there are styles, offset them as we have inserted into the strings
# offsets
if self.styles_offset != 0:
self.styles_offset += 4
chunk_data[24 : 24 + 4] = struct.pack("<I", self.strings_offset)
self.header.chunk_data = bytes(chunk_data)
self.dirty = True
return self.string_count - 1
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

View File

@@ -0,0 +1,951 @@
import argparse
import codecs
import errno
import numbers
import os
import platform
import re
import select
import shlex
import signal
import sys
import threading
import time
from types import FrameType
from typing import Any, Callable, List, Optional, Tuple, TypeVar, Union
import _frida
if platform.system() == "Windows":
import msvcrt
import colorama
import frida
from frida_tools.reactor import Reactor
AUX_OPTION_PATTERN = re.compile(r"(.+)=\((string|bool|int)\)(.+)")
T = TypeVar("T")
TargetType = Union[List[str], re.Pattern, int, str]
TargetTypeTuple = Tuple[str, TargetType]
def input_with_cancellable(cancellable: frida.Cancellable) -> str:
if platform.system() == "Windows":
result = ""
done = False
while not done:
while msvcrt.kbhit():
c = msvcrt.getwche()
if c in ("\x00", "\xe0"):
msvcrt.getwche()
continue
result += c
if c == "\n":
done = True
break
cancellable.raise_if_cancelled()
time.sleep(0.05)
return result
elif platform.system() in ["Darwin", "FreeBSD"]:
while True:
try:
rlist, _, _ = select.select([sys.stdin], [], [], 0.05)
except OSError as e:
if e.args[0] != errno.EINTR:
raise e
cancellable.raise_if_cancelled()
if sys.stdin in rlist:
return sys.stdin.readline()
else:
with cancellable.get_pollfd() as cancellable_fd:
try:
rlist, _, _ = select.select([sys.stdin, cancellable_fd], [], [])
except OSError as e:
if e.args[0] != errno.EINTR:
raise e
cancellable.raise_if_cancelled()
return sys.stdin.readline()
def await_enter(reactor: Reactor) -> None:
try:
input_with_cancellable(reactor.ui_cancellable)
except frida.OperationCancelledError:
pass
except KeyboardInterrupt:
print("")
def await_ctrl_c(reactor: Reactor) -> None:
while True:
try:
input_with_cancellable(reactor.ui_cancellable)
except frida.OperationCancelledError:
break
except KeyboardInterrupt:
break
def deserialize_relay(value: str) -> frida.Relay:
address, username, password, kind = value.split(",")
return frida.Relay(address, username, password, kind)
def create_target_parser(target_type: str) -> Callable[[str], TargetTypeTuple]:
def parse_target(value: str) -> TargetTypeTuple:
if target_type == "file":
return (target_type, [value])
if target_type == "gated":
return (target_type, re.compile(value))
if target_type == "pid":
return (target_type, int(value))
return (target_type, value)
return parse_target
class ConsoleState:
EMPTY = 1
STATUS = 2
TEXT = 3
class ConsoleApplication:
"""
ConsoleApplication is the base class for all of Frida tools, which contains
the common arguments of the tools. Each application can implement one or
more of several methods that can be inserted inside the flow of the
application.
The subclass should not expose any additional methods aside from __init__
and run methods that are defined by this class. These methods should not be
overridden without calling the super method.
"""
_target: Optional[TargetTypeTuple] = None
def __init__(
self,
run_until_return: Callable[["Reactor"], None] = await_enter,
on_stop: Optional[Callable[[], None]] = None,
args: Optional[List[str]] = None,
):
plain_terminal = os.environ.get("TERM", "").lower() == "none"
# Windows doesn't have SIGPIPE
if hasattr(signal, "SIGPIPE"):
signal.signal(signal.SIGPIPE, signal.SIG_DFL)
colorama.init(strip=True if plain_terminal else None)
parser = self._initialize_arguments_parser()
real_args = compute_real_args(parser, args=args)
options = parser.parse_args(real_args)
# handle scripts that don't need a target
if not hasattr(options, "args"):
options.args = []
self._initialize_device_arguments(parser, options)
self._initialize_target_arguments(parser, options)
self._reactor = Reactor(run_until_return, on_stop)
self._device: Optional[frida.core.Device] = None
self._schedule_on_output = lambda pid, fd, data: self._reactor.schedule(lambda: self._on_output(pid, fd, data))
self._schedule_on_device_lost = lambda: self._reactor.schedule(self._on_device_lost)
self._spawned_pid: Optional[int] = None
self._spawned_argv = None
self._selected_spawn: Optional[_frida.Spawn] = None
self._target_pid: Optional[int] = None
self._session: Optional[frida.core.Session] = None
self._schedule_on_session_detached = lambda reason, crash: self._reactor.schedule(
lambda: self._on_session_detached(reason, crash)
)
self._started = False
self._resumed = False
self._exit_status: Optional[int] = None
self._console_state = ConsoleState.EMPTY
self._have_terminal = sys.stdin.isatty() and sys.stdout.isatty() and not os.environ.get("TERM", "") == "dumb"
self._plain_terminal = plain_terminal
self._quiet = False
if sum(map(lambda v: int(v is not None), (self._device_id, self._device_type, self._host))) > 1:
parser.error("Only one of -D, -U, -R, and -H may be specified")
self._initialize_target(parser, options)
try:
self._initialize(parser, options, options.args)
except Exception as e:
parser.error(str(e))
def _initialize_device_arguments(self, parser: argparse.ArgumentParser, options: argparse.Namespace) -> None:
if self._needs_device():
self._device_id = options.device_id
self._device_type = options.device_type
self._host = options.host
if all([x is None for x in [self._device_id, self._device_type, self._host]]):
self._device_id = os.environ.get("FRIDA_DEVICE")
if self._device_id is None:
self._host = os.environ.get("FRIDA_HOST")
self._certificate = options.certificate or os.environ.get("FRIDA_CERTIFICATE")
self._origin = options.origin or os.environ.get("FRIDA_ORIGIN")
self._token = options.token or os.environ.get("FRIDA_TOKEN")
self._keepalive_interval = options.keepalive_interval
self._session_transport = options.session_transport
self._stun_server = options.stun_server
self._relays = options.relays
else:
self._device_id = None
self._device_type = None
self._host = None
self._certificate = None
self._origin = None
self._token = None
self._keepalive_interval = None
self._session_transport = "multiplexed"
self._stun_server = None
self._relays = None
def _initialize_target_arguments(self, parser: argparse.ArgumentParser, options: argparse.Namespace) -> None:
if self._needs_target():
self._stdio = options.stdio
self._aux = options.aux
self._realm = options.realm
self._runtime = options.runtime
self._enable_debugger = options.enable_debugger
self._squelch_crash = options.squelch_crash
else:
self._stdio = "inherit"
self._aux = []
self._realm = "native"
self._runtime = "qjs"
self._enable_debugger = False
self._squelch_crash = False
def _initialize_target(self, parser: argparse.ArgumentParser, options: argparse.Namespace) -> None:
if self._needs_target():
target = getattr(options, "target", None)
if target is None:
if len(options.args) < 1:
parser.error("target must be specified")
target = infer_target(options.args[0])
options.args.pop(0)
target = expand_target(target)
if target[0] == "file":
if not isinstance(target[1], list):
raise ValueError("file target must be a list of strings")
argv = target[1]
argv.extend(options.args)
options.args = []
self._target = target
else:
self._target = None
def _initialize_arguments_parser(self) -> argparse.ArgumentParser:
parser = self._initialize_base_arguments_parser()
self._add_options(parser)
return parser
def _initialize_base_arguments_parser(self) -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(usage=self._usage())
if self._needs_device():
self._add_device_arguments(parser)
if self._needs_target():
self._add_target_arguments(parser)
parser.add_argument(
"-O", "--options-file", help="text file containing additional command line options", metavar="FILE"
)
parser.add_argument("--version", action="version", version=frida.__version__)
return parser
def _add_device_arguments(self, parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"-D", "--device", help="connect to device with the given ID", metavar="ID", dest="device_id"
)
parser.add_argument(
"-U", "--usb", help="connect to USB device", action="store_const", const="usb", dest="device_type"
)
parser.add_argument(
"-R",
"--remote",
help="connect to remote frida-server",
action="store_const",
const="remote",
dest="device_type",
)
parser.add_argument("-H", "--host", help="connect to remote frida-server on HOST")
parser.add_argument("--certificate", help="speak TLS with HOST, expecting CERTIFICATE")
parser.add_argument("--origin", help="connect to remote server with “Origin” header set to ORIGIN")
parser.add_argument("--token", help="authenticate with HOST using TOKEN")
parser.add_argument(
"--keepalive-interval",
help="set keepalive interval in seconds, or 0 to disable (defaults to -1 to auto-select based on transport)",
metavar="INTERVAL",
type=int,
)
parser.add_argument(
"--p2p",
help="establish a peer-to-peer connection with target",
action="store_const",
const="p2p",
dest="session_transport",
default="multiplexed",
)
parser.add_argument("--stun-server", help="set STUN server ADDRESS to use with --p2p", metavar="ADDRESS")
parser.add_argument(
"--relay",
help="add relay to use with --p2p",
metavar="address,username,password,turn-{udp,tcp,tls}",
dest="relays",
action="append",
type=deserialize_relay,
)
def _add_target_arguments(self, parser: argparse.ArgumentParser) -> None:
parser.add_argument("-f", "--file", help="spawn FILE", dest="target", type=create_target_parser("file"))
parser.add_argument(
"-F",
"--attach-frontmost",
help="attach to frontmost application",
dest="target",
action="store_const",
const=("frontmost", None),
)
parser.add_argument(
"-n",
"--attach-name",
help="attach to NAME",
metavar="NAME",
dest="target",
type=create_target_parser("name"),
)
parser.add_argument(
"-N",
"--attach-identifier",
help="attach to IDENTIFIER",
metavar="IDENTIFIER",
dest="target",
type=create_target_parser("identifier"),
)
parser.add_argument(
"-p", "--attach-pid", help="attach to PID", metavar="PID", dest="target", type=create_target_parser("pid")
)
parser.add_argument(
"-W",
"--await",
help="await spawn matching PATTERN",
metavar="PATTERN",
dest="target",
type=create_target_parser("gated"),
)
parser.add_argument(
"--stdio",
help="stdio behavior when spawning (defaults to “inherit”)",
choices=["inherit", "pipe"],
default="inherit",
)
parser.add_argument(
"--aux",
help="set aux option when spawning, such as “uid=(int)42” (supported types are: string, bool, int)",
metavar="option",
action="append",
dest="aux",
default=[],
)
parser.add_argument("--realm", help="realm to attach in", choices=["native", "emulated"], default="native")
parser.add_argument("--runtime", help="script runtime to use", choices=["qjs", "v8"])
parser.add_argument(
"--debug",
help="enable the Node.js compatible script debugger",
action="store_true",
dest="enable_debugger",
default=False,
)
parser.add_argument(
"--squelch-crash",
help="if enabled, will not dump crash report to console",
action="store_true",
default=False,
)
parser.add_argument("args", help="extra arguments and/or target", nargs="*")
def run(self) -> None:
mgr = frida.get_device_manager()
on_devices_changed = lambda: self._reactor.schedule(self._try_start)
mgr.on("changed", on_devices_changed)
self._reactor.schedule(self._try_start)
self._reactor.schedule(self._show_message_if_no_device, delay=1)
signal.signal(signal.SIGTERM, self._on_sigterm)
self._reactor.run()
if self._started:
try:
self._perform_on_background_thread(self._stop)
except frida.OperationCancelledError:
pass
if self._session is not None:
self._session.off("detached", self._schedule_on_session_detached)
try:
self._perform_on_background_thread(self._session.detach)
except frida.OperationCancelledError:
pass
self._session = None
if self._device is not None:
self._device.off("output", self._schedule_on_output)
self._device.off("lost", self._schedule_on_device_lost)
mgr.off("changed", on_devices_changed)
frida.shutdown()
sys.exit(self._exit_status)
def _add_options(self, parser: argparse.ArgumentParser) -> None:
"""
override this method if you want to add custom arguments to your
command. The parser command is an argparse object, you should add the
options to him.
"""
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
"""
override this method if you need to have additional initialization code
before running, maybe to use your custom options from the `_add_options`
method.
"""
def _usage(self) -> str:
"""
override this method if to add a custom usage message
"""
return "%(prog)s [options]"
def _needs_device(self) -> bool:
"""
override this method if your command need to get a device from the user.
"""
return True
def _needs_target(self) -> bool:
"""
override this method if your command does not need to get a target
process from the user.
"""
return False
def _start(self) -> None:
"""
override this method with the logic of your command, it will run after
the class is fully initialized with a connected device/target if you
required one.
"""
def _stop(self) -> None:
"""
override this method if you have something you need to do at the end of
your command, maybe cleaning up some objects.
"""
def _resume(self) -> None:
if self._resumed:
return
if self._spawned_pid is not None:
assert self._device is not None
self._device.resume(self._spawned_pid)
assert self._target is not None
if self._target[0] == "gated":
self._device.disable_spawn_gating()
self._device.off("spawn-added", self._on_spawn_added)
self._resumed = True
def _exit(self, exit_status: int) -> None:
self._exit_status = exit_status
self._reactor.stop()
def _try_start(self) -> None:
if self._device is not None:
return
if self._device_id is not None:
try:
self._device = frida.get_device(self._device_id)
except:
self._update_status(f"Device '{self._device_id}' not found")
self._exit(1)
return
elif (self._host is not None) or (self._device_type == "remote"):
host = self._host
options = {}
if self._certificate is not None:
options["certificate"] = self._certificate
if self._origin is not None:
options["origin"] = self._origin
if self._token is not None:
options["token"] = self._token
if self._keepalive_interval is not None:
options["keepalive_interval"] = self._keepalive_interval
if host is None and len(options) == 0:
self._device = frida.get_remote_device()
else:
self._device = frida.get_device_manager().add_remote_device(
host if host is not None else "127.0.0.1", **options
)
elif self._device_type is not None:
self._device = find_device(self._device_type)
if self._device is None:
return
else:
self._device = frida.get_local_device()
self._on_device_found()
self._device.on("output", self._schedule_on_output)
self._device.on("lost", self._schedule_on_device_lost)
if self._target is not None:
target_type, target_value = self._target
if target_type == "gated":
self._device.on("spawn-added", self._on_spawn_added)
try:
self._device.enable_spawn_gating()
except Exception as e:
self._update_status(f"Failed to enable spawn gating: {e}")
self._exit(1)
return
self._update_status("Waiting for spawn to appear...")
return
spawning = True
try:
if target_type == "frontmost":
try:
app = self._device.get_frontmost_application()
except Exception as e:
self._update_status(f"Unable to get frontmost application on {self._device.name}: {e}")
self._exit(1)
return
if app is None:
self._update_status(f"No frontmost application on {self._device.name}")
self._exit(1)
return
self._target = ("name", app.name)
attach_target = app.pid
elif target_type == "identifier":
spawning = False
app_list = self._device.enumerate_applications()
app_identifier_lc = target_value.lower()
matching = [app for app in app_list if app.identifier.lower() == app_identifier_lc]
if len(matching) == 1 and matching[0].pid != 0:
attach_target = matching[0].pid
elif len(matching) > 1:
raise frida.ProcessNotFoundError(
"ambiguous identifier; it matches: %s"
% ", ".join([f"{process.identifier} (pid: {process.pid})" for process in matching])
)
else:
raise frida.ProcessNotFoundError("unable to find process with identifier '%s'" % target_value)
elif target_type == "file":
argv = target_value
if not self._quiet:
self._update_status(f"Spawning `{' '.join(argv)}`...")
aux_kwargs = {}
if self._aux is not None:
aux_kwargs = dict([parse_aux_option(o) for o in self._aux])
self._spawned_pid = self._device.spawn(argv, stdio=self._stdio, **aux_kwargs)
self._spawned_argv = argv
attach_target = self._spawned_pid
else:
attach_target = target_value
if not isinstance(attach_target, numbers.Number):
attach_target = self._device.get_process(attach_target).pid
if not self._quiet:
self._update_status("Attaching...")
spawning = False
self._attach(attach_target)
except frida.OperationCancelledError:
self._exit(0)
return
except Exception as e:
if spawning:
self._update_status(f"Failed to spawn: {e}")
else:
self._update_status(f"Failed to attach: {e}")
self._exit(1)
return
self._start()
self._started = True
def _attach(self, pid: int) -> None:
self._target_pid = pid
assert self._device is not None
self._session = self._device.attach(pid, realm=self._realm)
self._session.on("detached", self._schedule_on_session_detached)
if self._session_transport == "p2p":
peer_options = {}
if self._stun_server is not None:
peer_options["stun_server"] = self._stun_server
if self._relays is not None:
peer_options["relays"] = self._relays
self._session.setup_peer_connection(**peer_options)
def _on_script_created(self, script: frida.core.Script) -> None:
if self._enable_debugger:
script.enable_debugger()
self._print("Chrome Inspector server listening on port 9229\n")
def _show_message_if_no_device(self) -> None:
if self._device is None:
self._print("Waiting for USB device to appear...")
def _on_sigterm(self, n: int, f: Optional[FrameType]) -> None:
self._reactor.cancel_io()
self._exit(0)
def _on_spawn_added(self, spawn: _frida.Spawn) -> None:
thread = threading.Thread(target=self._handle_spawn, args=(spawn,))
thread.start()
def _handle_spawn(self, spawn: _frida.Spawn) -> None:
pid = spawn.pid
pattern = self._target[1]
if pattern.match(spawn.identifier) is None or self._selected_spawn is not None:
self._print(
colorama.Fore.YELLOW + colorama.Style.BRIGHT + "Ignoring: " + str(spawn) + colorama.Style.RESET_ALL
)
try:
if self._device is not None:
self._device.resume(pid)
except:
pass
return
self._selected_spawn = spawn
self._print(colorama.Fore.GREEN + colorama.Style.BRIGHT + "Handling: " + str(spawn) + colorama.Style.RESET_ALL)
try:
self._attach(pid)
self._reactor.schedule(lambda: self._on_spawn_handled(spawn))
except Exception as e:
error = e
self._reactor.schedule(lambda: self._on_spawn_unhandled(spawn, error))
def _on_spawn_handled(self, spawn: _frida.Spawn) -> None:
self._spawned_pid = spawn.pid
self._start()
self._started = True
def _on_spawn_unhandled(self, spawn: _frida.Spawn, error: Exception) -> None:
self._update_status(f"Failed to handle spawn: {error}")
self._exit(1)
def _on_output(self, pid: int, fd: int, data: Optional[bytes]) -> None:
if pid != self._target_pid or data is None:
return
if fd == 1:
prefix = "stdout> "
stream = sys.stdout
else:
prefix = "stderr> "
stream = sys.stderr
encoding = stream.encoding or "UTF-8"
text = data.decode(encoding, errors="replace")
if text.endswith("\n"):
text = text[:-1]
lines = text.split("\n")
self._print(prefix + ("\n" + prefix).join(lines))
def _on_device_found(self) -> None:
pass
def _on_device_lost(self) -> None:
if self._exit_status is not None:
return
self._print("Device disconnected.")
self._exit(1)
def _on_session_detached(self, reason: str, crash) -> None:
if crash is None:
message = reason[0].upper() + reason[1:].replace("-", " ")
else:
message = "Process crashed: " + crash.summary
self._print(colorama.Fore.RED + colorama.Style.BRIGHT + message + colorama.Style.RESET_ALL)
if crash is not None:
if self._squelch_crash is True:
self._print("\n*** Crash report was squelched due to user setting. ***")
else:
self._print("\n***\n{}\n***".format(crash.report.rstrip("\n")))
self._exit(1)
def _clear_status(self) -> None:
if self._console_state == ConsoleState.STATUS:
print(colorama.Cursor.UP() + (80 * " "))
def _update_status(self, message: str) -> None:
if self._have_terminal:
if self._console_state == ConsoleState.STATUS:
cursor_position = colorama.Cursor.UP()
else:
cursor_position = ""
print("%-80s" % (cursor_position + colorama.Style.BRIGHT + message + colorama.Style.RESET_ALL,))
self._console_state = ConsoleState.STATUS
else:
print(colorama.Style.BRIGHT + message + colorama.Style.RESET_ALL)
def _print(self, *args: Any, **kwargs: Any) -> None:
encoded_args: List[Any] = []
encoding = sys.stdout.encoding or "UTF-8"
if encoding == "UTF-8":
encoded_args = list(args)
else:
for arg in args:
if isinstance(arg, str):
encoded_args.append(arg.encode(encoding, errors="backslashreplace").decode(encoding))
else:
encoded_args.append(arg)
print(*encoded_args, **kwargs)
self._console_state = ConsoleState.TEXT
def _log(self, level: str, text: str) -> None:
if level == "info":
self._print(text)
else:
color = colorama.Fore.RED if level == "error" else colorama.Fore.YELLOW
text = color + colorama.Style.BRIGHT + text + colorama.Style.RESET_ALL
if level == "error":
self._print(text, file=sys.stderr)
else:
self._print(text)
def _perform_on_reactor_thread(self, f: Callable[[], T]) -> T:
completed = threading.Event()
result = [None, None]
def work() -> None:
try:
result[0] = f()
except Exception as e:
result[1] = e
completed.set()
self._reactor.schedule(work)
while not completed.is_set():
try:
completed.wait()
except KeyboardInterrupt:
self._reactor.cancel_io()
continue
error = result[1]
if error is not None:
raise error
return result[0]
def _perform_on_background_thread(self, f: Callable[[], T], timeout: Optional[float] = None) -> T:
result = [None, None]
def work() -> None:
with self._reactor.io_cancellable:
try:
result[0] = f()
except Exception as e:
result[1] = e
worker = threading.Thread(target=work)
worker.start()
try:
worker.join(timeout)
except KeyboardInterrupt:
self._reactor.cancel_io()
if timeout is not None and worker.is_alive():
self._reactor.cancel_io()
while worker.is_alive():
try:
worker.join()
except KeyboardInterrupt:
pass
error = result[1]
if error is not None:
raise error
return result[0]
def _get_default_frida_dir(self) -> str:
return os.path.join(os.path.expanduser("~"), ".frida")
def _get_windows_frida_dir(self) -> str:
appdata = os.environ["LOCALAPPDATA"]
return os.path.join(appdata, "frida")
def _get_or_create_config_dir(self) -> str:
config_dir = os.path.join(self._get_default_frida_dir(), "config")
if platform.system() == "Linux":
xdg_config_home = os.getenv("XDG_CONFIG_HOME", os.path.expanduser("~/.config"))
config_dir = os.path.join(xdg_config_home, "frida")
elif platform.system() == "Windows":
config_dir = os.path.join(self._get_windows_frida_dir(), "Config")
if not os.path.exists(config_dir):
os.makedirs(config_dir)
return config_dir
def _get_or_create_data_dir(self) -> str:
data_dir = os.path.join(self._get_default_frida_dir(), "data")
if platform.system() == "Linux":
xdg_data_home = os.getenv("XDG_DATA_HOME", os.path.expanduser("~/.local/share"))
data_dir = os.path.join(xdg_data_home, "frida")
elif platform.system() == "Windows":
data_dir = os.path.join(self._get_windows_frida_dir(), "Data")
if not os.path.exists(data_dir):
os.makedirs(data_dir)
return data_dir
def _get_or_create_state_dir(self) -> str:
state_dir = os.path.join(self._get_default_frida_dir(), "state")
if platform.system() == "Linux":
xdg_state_home = os.getenv("XDG_STATE_HOME", os.path.expanduser("~/.local/state"))
state_dir = os.path.join(xdg_state_home, "frida")
elif platform.system() == "Windows":
appdata = os.environ["LOCALAPPDATA"]
state_dir = os.path.join(appdata, "frida", "State")
if not os.path.exists(state_dir):
os.makedirs(state_dir)
return state_dir
def compute_real_args(parser: argparse.ArgumentParser, args: Optional[List[str]] = None) -> List[str]:
if args is None:
args = sys.argv[1:]
real_args = normalize_options_file_args(args)
files_processed = set()
while True:
offset = find_options_file_offset(real_args, parser)
if offset == -1:
break
file_path = os.path.abspath(real_args[offset + 1])
if file_path in files_processed:
parser.error(f"File '{file_path}' given twice as -O argument")
if os.path.isfile(file_path):
with codecs.open(file_path, "r", "utf-8") as f:
new_arg_text = f.read()
else:
parser.error(f"File '{file_path}' following -O option is not a valid file")
real_args = insert_options_file_args_in_list(real_args, offset, new_arg_text)
files_processed.add(file_path)
return real_args
def normalize_options_file_args(raw_args: List[str]) -> List[str]:
result = []
for arg in raw_args:
if arg.startswith("--options-file="):
result.append(arg[0:14])
result.append(arg[15:])
else:
result.append(arg)
return result
def find_options_file_offset(arglist: List[str], parser: argparse.ArgumentParser) -> int:
for i, arg in enumerate(arglist):
if arg in ("-O", "--options-file"):
if i < len(arglist) - 1:
return i
else:
parser.error("No argument given for -O option")
return -1
def insert_options_file_args_in_list(args: List[str], offset: int, new_arg_text: str) -> List[str]:
new_args = shlex.split(new_arg_text)
new_args = normalize_options_file_args(new_args)
new_args_list = args[:offset] + new_args + args[offset + 2 :]
return new_args_list
def find_device(device_type: str) -> Optional[frida.core.Device]:
for device in frida.enumerate_devices():
if device.type == device_type:
return device
return None
def infer_target(target_value: str) -> TargetTypeTuple:
if (
target_value.startswith(".")
or target_value.startswith(os.path.sep)
or (
platform.system() == "Windows"
and target_value[0].isalpha()
and target_value[1] == ":"
and target_value[2] == "\\"
)
):
return ("file", [target_value])
try:
return ("pid", int(target_value))
except:
pass
return ("name", target_value)
def expand_target(target: TargetTypeTuple) -> TargetTypeTuple:
target_type, target_value = target
if target_type == "file" and isinstance(target_value, list):
target_value = [target_value[0]]
return (target_type, target_value)
def parse_aux_option(option: str) -> Tuple[str, Union[str, bool, int]]:
m = AUX_OPTION_PATTERN.match(option)
if m is None:
raise ValueError("expected name=(type)value, e.g. “uid=(int)42”; supported types are: string, bool, int")
name = m.group(1)
type_decl = m.group(2)
raw_value = m.group(3)
if type_decl == "string":
value = raw_value
elif type_decl == "bool":
value = bool(raw_value)
else:
value = int(raw_value)
return (name, value)

View File

@@ -0,0 +1,62 @@
from typing import Any, Dict, Union
from colorama import Fore, Style
STYLE_FILE = Fore.CYAN + Style.BRIGHT
STYLE_LOCATION = Fore.LIGHTYELLOW_EX
STYLE_ERROR = Fore.RED + Style.BRIGHT
STYLE_WARNING = Fore.YELLOW + Style.BRIGHT
STYLE_CODE = Fore.WHITE + Style.DIM
STYLE_RESET_ALL = Style.RESET_ALL
CATEGORY_STYLE = {
"warning": STYLE_WARNING,
"error": STYLE_ERROR,
}
def format_error(error: BaseException) -> str:
return STYLE_ERROR + str(error) + Style.RESET_ALL
def format_compiling(script_path: str, cwd: str) -> str:
name = format_filename(script_path, cwd)
return f"{STYLE_RESET_ALL}Compiling {STYLE_FILE}{name}{STYLE_RESET_ALL}..."
def format_compiled(
script_path: str, cwd: str, time_started: Union[int, float], time_finished: Union[int, float]
) -> str:
name = format_filename(script_path, cwd)
elapsed = int((time_finished - time_started) * 1000.0)
return f"{STYLE_RESET_ALL}Compiled {STYLE_FILE}{name}{STYLE_RESET_ALL}{STYLE_CODE} ({elapsed} ms){STYLE_RESET_ALL}"
def format_diagnostic(diag: Dict[str, Any], cwd: str) -> str:
category = diag["category"]
code = diag["code"]
text = diag["text"]
file = diag.get("file", None)
if file is not None:
filename = format_filename(file["path"], cwd)
line = file["line"] + 1
character = file["character"] + 1
path_segment = f"{STYLE_FILE}{filename}{STYLE_RESET_ALL}"
line_segment = f"{STYLE_LOCATION}{line}{STYLE_RESET_ALL}"
character_segment = f"{STYLE_LOCATION}{character}{STYLE_RESET_ALL}"
prefix = f"{path_segment}:{line_segment}:{character_segment} - "
else:
prefix = ""
category_style = CATEGORY_STYLE.get(category, STYLE_RESET_ALL)
return f"{prefix}{category_style}{category}{STYLE_RESET_ALL} {STYLE_CODE}TS{code}{STYLE_RESET_ALL}: {text}"
def format_filename(path: str, cwd: str) -> str:
if path.startswith(cwd):
return path[len(cwd) + 1 :]
return path

View File

@@ -0,0 +1,117 @@
import argparse
import os
import sys
from timeit import default_timer as timer
from typing import Any, Dict, List, Optional
import frida
from frida_tools.application import ConsoleApplication, await_ctrl_c
from frida_tools.cli_formatting import format_compiled, format_compiling, format_diagnostic, format_error
def main() -> None:
app = CompilerApplication()
app.run()
class CompilerApplication(ConsoleApplication):
def __init__(self) -> None:
super().__init__(await_ctrl_c)
def _usage(self) -> str:
return "%(prog)s [options] <module>"
def _add_options(self, parser: argparse.ArgumentParser) -> None:
parser.add_argument("module", help="TypeScript/JavaScript module to compile")
parser.add_argument("-o", "--output", help="write output to <file>")
parser.add_argument("-w", "--watch", help="watch for changes and recompile", action="store_true")
parser.add_argument("-S", "--no-source-maps", help="omit source-maps", action="store_true")
parser.add_argument("-c", "--compress", help="compress using terser", action="store_true")
parser.add_argument("-v", "--verbose", help="be verbose", action="store_true")
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
self._module = os.path.abspath(options.module)
self._output = options.output
self._mode = "watch" if options.watch else "build"
self._verbose = self._mode == "watch" or options.verbose
self._compiler_options = {
"project_root": os.getcwd(),
"source_maps": "omitted" if options.no_source_maps else "included",
"compression": "terser" if options.compress else "none",
}
compiler = frida.Compiler()
self._compiler = compiler
def on_compiler_finished() -> None:
self._reactor.schedule(lambda: self._on_compiler_finished())
def on_compiler_output(bundle: str) -> None:
self._reactor.schedule(lambda: self._on_compiler_output(bundle))
def on_compiler_diagnostics(diagnostics: List[Dict[str, Any]]) -> None:
self._reactor.schedule(lambda: self._on_compiler_diagnostics(diagnostics))
compiler.on("starting", self._on_compiler_starting)
compiler.on("finished", on_compiler_finished)
compiler.on("output", on_compiler_output)
compiler.on("diagnostics", on_compiler_diagnostics)
self._compilation_started: Optional[float] = None
def _needs_device(self) -> bool:
return False
def _start(self) -> None:
try:
if self._mode == "build":
self._compiler.build(self._module, **self._compiler_options)
self._exit(0)
else:
self._compiler.watch(self._module, **self._compiler_options)
except Exception as e:
error = e
self._reactor.schedule(lambda: self._on_fatal_error(error))
def _on_fatal_error(self, error: Exception) -> None:
self._print(format_error(error))
self._exit(1)
def _on_compiler_starting(self) -> None:
self._compilation_started = timer()
if self._verbose:
self._reactor.schedule(lambda: self._print_compiler_starting())
def _print_compiler_starting(self) -> None:
if self._mode == "watch":
sys.stdout.write("\x1Bc")
self._print(format_compiling(self._module, os.getcwd()))
def _on_compiler_finished(self) -> None:
if self._verbose:
time_finished = timer()
assert self._compilation_started is not None
self._print(format_compiled(self._module, os.getcwd(), self._compilation_started, time_finished))
def _on_compiler_output(self, bundle: str) -> None:
if self._output is not None:
try:
with open(self._output, "w", encoding="utf-8", newline="\n") as f:
f.write(bundle)
except Exception as e:
self._on_fatal_error(e)
else:
sys.stdout.write(bundle)
def _on_compiler_diagnostics(self, diagnostics: List[Dict[str, Any]]) -> None:
cwd = os.getcwd()
for diag in diagnostics:
self._print(format_diagnostic(diag, cwd))
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

View File

@@ -0,0 +1,265 @@
import argparse
import codecs
import os
import platform
from typing import Dict, List, Tuple
import frida
from frida_tools.application import ConsoleApplication
def main() -> None:
app = CreatorApplication()
app.run()
class CreatorApplication(ConsoleApplication):
def _usage(self) -> str:
return "%(prog)s [options] -t agent|cmodule"
def _add_options(self, parser: argparse.ArgumentParser) -> None:
default_project_name = os.path.basename(os.getcwd())
parser.add_argument(
"-n", "--project-name", help="project name", dest="project_name", default=default_project_name
)
parser.add_argument("-o", "--output-directory", help="output directory", dest="outdir", default=".")
parser.add_argument("-t", "--template", help="template file: cmodule|agent", dest="template", default=None)
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
parsed_args = parser.parse_args()
if not parsed_args.template:
parser.error("template must be specified")
impl = getattr(self, "_generate_" + parsed_args.template, None)
if impl is None:
parser.error("unknown template type")
self._generate = impl
self._project_name = options.project_name
self._outdir = options.outdir
def _needs_device(self) -> bool:
return False
def _start(self) -> None:
(assets, message) = self._generate()
outdir = self._outdir
for name, data in assets.items():
asset_path = os.path.join(outdir, name)
asset_dir = os.path.dirname(asset_path)
try:
os.makedirs(asset_dir)
except:
pass
with codecs.open(asset_path, "wb", "utf-8") as f:
f.write(data)
self._print("Created", asset_path)
self._print("\n" + message)
self._exit(0)
def _generate_agent(self) -> Tuple[Dict[str, str], str]:
assets = {}
assets[
"package.json"
] = f"""{{
"name": "{self._project_name}-agent",
"version": "1.0.0",
"description": "Frida agent written in TypeScript",
"private": true,
"main": "agent/index.ts",
"scripts": {{
"prepare": "npm run build",
"build": "frida-compile agent/index.ts -o _agent.js -c",
"watch": "frida-compile agent/index.ts -o _agent.js -w"
}},
"devDependencies": {{
"@types/frida-gum": "^18.3.1",
"@types/node": "^18.14.0",
"frida-compile": "^16.1.8"
}}
}}
"""
assets[
"tsconfig.json"
] = """\
{
"compilerOptions": {
"target": "es2020",
"lib": ["es2020"],
"allowJs": true,
"noEmit": true,
"strict": true,
"esModuleInterop": true,
"moduleResolution": "node16"
}
}
"""
assets[
"agent/index.ts"
] = """\
import { log } from "./logger.js";
const header = Memory.alloc(16);
header
.writeU32(0xdeadbeef).add(4)
.writeU32(0xd00ff00d).add(4)
.writeU64(uint64("0x1122334455667788"));
log(hexdump(header.readByteArray(16) as ArrayBuffer, { ansi: true }));
Process.getModuleByName("libSystem.B.dylib")
.enumerateExports()
.slice(0, 16)
.forEach((exp, index) => {
log(`export ${index}: ${exp.name}`);
});
Interceptor.attach(Module.getExportByName(null, "open"), {
onEnter(args) {
const path = args[0].readUtf8String();
log(`open() path="${path}"`);
}
});
"""
assets[
"agent/logger.ts"
] = """\
export function log(message: string): void {
console.log(message);
}
"""
assets[".gitignore"] = "/node_modules/\n"
message = """\
Run `npm install` to bootstrap, then:
- Keep one terminal running: npm run watch
- Inject agent using the REPL: frida Calculator -l _agent.js
- Edit agent/*.ts - REPL will live-reload on save
Tip: Use an editor like Visual Studio Code for code completion, inline docs,
instant type-checking feedback, refactoring tools, etc.
"""
return (assets, message)
def _generate_cmodule(self) -> Tuple[Dict[str, str], str]:
assets = {}
assets[
"meson.build"
] = f"""\
project('{self._project_name}', 'c',
default_options: 'buildtype=release',
)
shared_module('{self._project_name}', '{self._project_name}.c',
name_prefix: '',
include_directories: include_directories('include'),
)
"""
assets[
self._project_name + ".c"
] = """\
#include <gum/guminterceptor.h>
static void frida_log (const char * format, ...);
extern void _frida_log (const gchar * message);
void
init (void)
{
frida_log ("init()");
}
void
finalize (void)
{
frida_log ("finalize()");
}
void
on_enter (GumInvocationContext * ic)
{
gpointer arg0;
arg0 = gum_invocation_context_get_nth_argument (ic, 0);
frida_log ("on_enter() arg0=%p", arg0);
}
void
on_leave (GumInvocationContext * ic)
{
gpointer retval;
retval = gum_invocation_context_get_return_value (ic);
frida_log ("on_leave() retval=%p", retval);
}
static void
frida_log (const char * format,
...)
{
gchar * message;
va_list args;
va_start (args, format);
message = g_strdup_vprintf (format, args);
va_end (args);
_frida_log (message);
g_free (message);
}
"""
assets[".gitignore"] = "/build/\n"
session = frida.attach(0)
script = session.create_script("rpc.exports.getBuiltins = () => CModule.builtins;")
self._on_script_created(script)
script.load()
builtins = script.exports_sync.get_builtins()
script.unload()
session.detach()
for name, data in builtins["headers"].items():
assets["include/" + name] = data
system = platform.system()
if system == "Windows":
module_extension = "dll"
elif system == "Darwin":
module_extension = "dylib"
else:
module_extension = "so"
cmodule_path = os.path.join(self._outdir, "build", self._project_name + "." + module_extension)
message = f"""\
Run `meson build && ninja -C build` to build, then:
- Inject CModule using the REPL: frida Calculator -C {cmodule_path}
- Edit *.c, and build incrementally through `ninja -C build`
- REPL will live-reload whenever {cmodule_path} changes on disk
"""
return (assets, message)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

View File

@@ -0,0 +1,239 @@
import argparse
import threading
from typing import List, Mapping, Optional, Tuple
import frida
from frida_tools.application import ConsoleApplication, await_enter
from frida_tools.model import Function, Module, ModuleFunction
from frida_tools.reactor import Reactor
class UI:
def on_sample_start(self, total: int) -> None:
pass
def on_sample_result(
self,
module_functions: Mapping[Module, List[Tuple[ModuleFunction, int]]],
dynamic_functions: List[Tuple[ModuleFunction, int]],
) -> None:
pass
def _on_script_created(self, script: frida.core.Script) -> None:
pass
class Discoverer:
def __init__(self, reactor: Reactor) -> None:
self._reactor = reactor
self._ui = None
self._script: Optional[frida.core.Script] = None
def dispose(self) -> None:
if self._script is not None:
try:
self._script.unload()
except:
pass
self._script = None
def start(self, session: frida.core.Session, runtime: str, ui: UI) -> None:
def on_message(message, data) -> None:
print(message, data)
self._ui = ui
script = session.create_script(name="discoverer", source=self._create_discover_script(), runtime=runtime)
self._script = script
self._ui._on_script_created(script)
script.on("message", on_message)
script.load()
params = script.exports_sync.start()
ui.on_sample_start(params["total"])
def stop(self) -> None:
result = self._script.exports_sync.stop()
modules = {
int(module_id): Module(m["name"], int(m["base"], 16), m["size"], m["path"])
for module_id, m in result["modules"].items()
}
module_functions = {}
dynamic_functions = []
for module_id, name, visibility, raw_address, count in result["targets"]:
address = int(raw_address, 16)
if module_id != 0:
module = modules[module_id]
exported = visibility == "e"
function = ModuleFunction(module, name, address - module.base_address, exported)
functions = module_functions.get(module, [])
if len(functions) == 0:
module_functions[module] = functions
functions.append((function, count))
else:
function = Function(name, address)
dynamic_functions.append((function, count))
self._ui.on_sample_result(module_functions, dynamic_functions)
def _create_discover_script(self) -> str:
return """\
const threadIds = new Set();
const result = new Map();
rpc.exports = {
start: function () {
for (const { id: threadId } of Process.enumerateThreads()) {
threadIds.add(threadId);
Stalker.follow(threadId, {
events: { call: true },
onCallSummary(summary) {
for (const [address, count] of Object.entries(summary)) {
result.set(address, (result.get(address) ?? 0) + count);
}
}
});
}
return {
total: threadIds.size
};
},
stop: function () {
for (const threadId of threadIds.values()) {
Stalker.unfollow(threadId);
}
threadIds.clear();
const targets = [];
const modules = {};
const moduleMap = new ModuleMap();
const allModules = moduleMap.values().reduce((m, module) => m.set(module.path, module), new Map());
const moduleDetails = new Map();
let nextModuleId = 1;
for (const [address, count] of result.entries()) {
let moduleId = 0;
let name;
let visibility = 'i';
const addressPtr = ptr(address);
const path = moduleMap.findPath(addressPtr);
if (path !== null) {
const module = allModules.get(path);
let details = moduleDetails.get(path);
if (details !== undefined) {
moduleId = details.id;
} else {
moduleId = nextModuleId++;
details = {
id: moduleId,
exports: module.enumerateExports().reduce((m, e) => m.set(e.address.toString(), e.name), new Map())
};
moduleDetails.set(path, details);
modules[moduleId] = module;
}
const exportName = details.exports.get(address);
if (exportName !== undefined) {
name = exportName;
visibility = 'e';
} else {
name = 'sub_' + addressPtr.sub(module.base).toString(16);
}
} else {
name = 'dsub_' + addressPtr.toString(16);
}
targets.push([moduleId, name, visibility, address, count]);
}
result.clear();
return {
targets,
modules
};
}
};
"""
class DiscovererApplication(ConsoleApplication, UI):
_discoverer: Optional[Discoverer]
def __init__(self) -> None:
self._results_received = threading.Event()
ConsoleApplication.__init__(self, self._await_keys)
def _await_keys(self, reactor: Reactor) -> None:
await_enter(reactor)
reactor.schedule(lambda: self._discoverer.stop())
while reactor.is_running() and not self._results_received.is_set():
self._results_received.wait(0.5)
def _usage(self) -> str:
return "%(prog)s [options] target"
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
self._discoverer = None
def _needs_target(self) -> bool:
return True
def _start(self) -> None:
self._update_status("Injecting script...")
self._discoverer = Discoverer(self._reactor)
self._discoverer.start(self._session, self._runtime, self)
def _stop(self) -> None:
self._print("Stopping...")
assert self._discoverer is not None
self._discoverer.dispose()
self._discoverer = None
def on_sample_start(self, total: int) -> None:
self._update_status(f"Tracing {total} threads. Press ENTER to stop.")
self._resume()
def on_sample_result(
self,
module_functions: Mapping[Module, List[Tuple[ModuleFunction, int]]],
dynamic_functions: List[Tuple[ModuleFunction, int]],
) -> None:
for module, functions in module_functions.items():
self._print(module.name)
self._print("\t%-10s\t%s" % ("Calls", "Function"))
for function, count in sorted(functions, key=lambda item: item[1], reverse=True):
self._print("\t%-10d\t%s" % (count, function))
self._print("")
if len(dynamic_functions) > 0:
self._print("Dynamic functions:")
self._print("\t%-10s\t%s" % ("Calls", "Function"))
for function, count in sorted(dynamic_functions, key=lambda item: item[1], reverse=True):
self._print("\t%-10d\t%s" % (count, function))
self._results_received.set()
def main() -> None:
app = DiscovererApplication()
app.run()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,478 @@
import json
import os
import struct
from abc import ABC, abstractmethod
from pathlib import Path
from typing import List, Optional, Sequence, Tuple, TypeVar, Union
import frida
from frida.core import RPCException
from frida_tools.reactor import Reactor
CodeLocation = Union[
Tuple[str, str],
Tuple[str, Tuple[str, str]],
Tuple[str, Tuple[str, int]],
]
TraceThreadStrategy = Tuple[str, Tuple[str, int]]
TraceRangeStrategy = Tuple[str, Tuple[CodeLocation, Optional[CodeLocation]]]
TraceStrategy = Union[TraceThreadStrategy, TraceRangeStrategy]
def main() -> None:
import argparse
import threading
from prompt_toolkit import PromptSession, prompt
from prompt_toolkit.application import Application
from prompt_toolkit.formatted_text import AnyFormattedText, FormattedText
from prompt_toolkit.key_binding.defaults import load_key_bindings
from prompt_toolkit.key_binding.key_bindings import KeyBindings, merge_key_bindings
from prompt_toolkit.layout import Layout
from prompt_toolkit.layout.containers import HSplit
from prompt_toolkit.styles import BaseStyle
from prompt_toolkit.widgets import Label, RadioList
from frida_tools.application import ConsoleApplication
class InstructionTracerApplication(ConsoleApplication, InstructionTracerUI):
_itracer: Optional[InstructionTracer]
def __init__(self) -> None:
self._state = "starting"
self._ready = threading.Event()
self._cli = PromptSession()
super().__init__(self._process_input)
def _add_options(self, parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"-t", "--thread-id", help="trace THREAD_ID", metavar="THREAD_ID", dest="strategy", type=parse_thread_id
)
parser.add_argument(
"-i",
"--thread-index",
help="trace THREAD_INDEX",
metavar="THREAD_INDEX",
dest="strategy",
type=parse_thread_index,
)
parser.add_argument(
"-r",
"--range",
help="trace RANGE, e.g.: 0x1000..0x1008, libc.so!sleep, libc.so!0x1234, recv..memcpy",
metavar="RANGE",
dest="strategy",
type=parse_range,
)
parser.add_argument("-o", "--output", help="output to file", dest="outpath")
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
self._itracer = None
self._strategy = options.strategy
self._outpath = options.outpath
def _usage(self) -> str:
return "%(prog)s [options] target"
def _needs_target(self) -> bool:
return True
def _start(self) -> None:
self._update_status("Injecting script...")
self._itracer = InstructionTracer(self._reactor)
self._itracer.start(self._device, self._session, self._runtime, self)
self._ready.set()
def _stop(self) -> None:
assert self._itracer is not None
self._itracer.dispose()
self._itracer = None
try:
self._cli.app.exit()
except:
pass
def _process_input(self, reactor: Reactor) -> None:
try:
while self._ready.wait(0.5) != True:
if not reactor.is_running():
return
except KeyboardInterrupt:
reactor.cancel_io()
return
if self._state != "started":
return
try:
self._cli.prompt()
except:
pass
def get_trace_strategy(self) -> Optional[TraceStrategy]:
return self._strategy
def prompt_for_trace_strategy(self, threads: List[dict]) -> Optional[TraceStrategy]:
kind = radiolist_prompt(
title="Tracing strategy:",
values=[
("thread", "Thread"),
("range", "Range"),
],
)
if kind is None:
raise KeyboardInterrupt
if kind == "thread":
thread_id = radiolist_prompt(
title="Running threads:", values=[(t["id"], json.dumps(t)) for t in threads]
)
if thread_id is None:
raise KeyboardInterrupt
return ("thread", ("id", thread_id))
while True:
try:
text = prompt("Start address: ").strip()
if len(text) == 0:
continue
start = parse_code_location(text)
break
except Exception as e:
print(str(e))
continue
while True:
try:
text = prompt("End address (optional): ").strip()
if len(text) > 0:
end = parse_code_location(text)
else:
end = None
break
except Exception as e:
print(str(e))
continue
return ("range", (start, end))
def get_trace_output_path(self, suggested_name: Optional[str] = None) -> os.PathLike:
return self._outpath
def prompt_for_trace_output_path(self, suggested_name: str) -> Optional[os.PathLike]:
while True:
outpath = prompt("Output filename: ", default=suggested_name).strip()
if len(outpath) != 0:
break
return outpath
def on_trace_started(self) -> None:
self._state = "started"
def on_trace_stopped(self, error_message: Optional[str] = None) -> None:
self._state = "stopping"
if error_message is not None:
self._log(level="error", text=error_message)
self._exit(1)
else:
self._exit(0)
try:
self._cli.app.exit()
except:
pass
def on_trace_progress(self, total_blocks: int, total_bytes: int) -> None:
blocks_suffix = "s" if total_blocks != 1 else ""
self._cli.message = FormattedText(
[
("bold", "Tracing!"),
("", " Collected "),
("fg:green bold", human_readable_size(total_bytes)),
("", f" from {total_blocks} basic block{blocks_suffix}"),
]
)
self._cli.app.invalidate()
def parse_thread_id(value: str) -> TraceThreadStrategy:
return ("thread", ("id", int(value)))
def parse_thread_index(value: str) -> TraceThreadStrategy:
return ("thread", ("index", int(value)))
def parse_range(value: str) -> TraceRangeStrategy:
tokens = value.split("..", 1)
start = tokens[0]
end = tokens[1] if len(tokens) == 2 else None
return ("range", (parse_code_location(start), parse_code_location(end)))
def parse_code_location(value: Optional[str]) -> CodeLocation:
if value is None:
return None
if value.startswith("0x"):
return ("address", value)
tokens = value.split("!", 1)
if len(tokens) == 2:
name = tokens[0]
subval = tokens[1]
if subval.startswith("0x"):
return ("module-offset", (name, int(subval, 16)))
return ("module-export", (name, subval))
return ("symbol", tokens[0])
# Based on https://stackoverflow.com/a/43690506
def human_readable_size(size):
for unit in ["B", "KiB", "MiB", "GiB"]:
if size < 1024.0 or unit == "GiB":
break
size /= 1024.0
return f"{size:.2f} {unit}"
T = TypeVar("T")
# Based on https://github.com/prompt-toolkit/python-prompt-toolkit/issues/756#issuecomment-1294742392
def radiolist_prompt(
title: str = "",
values: Sequence[Tuple[T, AnyFormattedText]] = None,
default: Optional[T] = None,
cancel_value: Optional[T] = None,
style: Optional[BaseStyle] = None,
) -> T:
radio_list = RadioList(values, default)
radio_list.control.key_bindings.remove("enter")
bindings = KeyBindings()
@bindings.add("enter")
def exit_with_value(event):
radio_list._handle_enter()
event.app.exit(result=radio_list.current_value)
@bindings.add("c-c")
def backup_exit_with_value(event):
event.app.exit(result=cancel_value)
application = Application(
layout=Layout(HSplit([Label(title), radio_list])),
key_bindings=merge_key_bindings([load_key_bindings(), bindings]),
mouse_support=True,
style=style,
full_screen=False,
)
return application.run()
app = InstructionTracerApplication()
app.run()
class InstructionTracerUI(ABC):
@abstractmethod
def get_trace_strategy(self) -> Optional[TraceStrategy]:
raise NotImplementedError
def prompt_for_trace_strategy(self, threads: List[dict]) -> Optional[TraceStrategy]:
return None
@abstractmethod
def get_trace_output_path(self) -> Optional[os.PathLike]:
raise NotImplementedError
def prompt_for_trace_output_path(self, suggested_name: str) -> Optional[os.PathLike]:
return None
@abstractmethod
def on_trace_started(self) -> None:
raise NotImplementedError
@abstractmethod
def on_trace_stopped(self, error_message: Optional[str] = None) -> None:
raise NotImplementedError
def on_trace_progress(self, total_blocks: int, total_bytes: int) -> None:
pass
def _on_script_created(self, script: frida.core.Script) -> None:
pass
class InstructionTracer:
FILE_MAGIC = b"ITRC"
def __init__(self, reactor: Reactor) -> None:
self._reactor = reactor
self._outfile = None
self._ui: Optional[InstructionTracerUI] = None
self._total_blocks = 0
self._tracer_script: Optional[frida.core.Script] = None
self._reader_script: Optional[frida.core.Script] = None
self._reader_api = None
def dispose(self) -> None:
if self._reader_api is not None:
try:
self._reader_api.stop_buffer_reader()
except:
pass
self._reader_api = None
if self._reader_script is not None:
try:
self._reader_script.unload()
except:
pass
self._reader_script = None
if self._tracer_script is not None:
try:
self._tracer_script.unload()
except:
pass
self._tracer_script = None
def start(
self, device: frida.core.Device, session: frida.core.Session, runtime: str, ui: InstructionTracerUI
) -> None:
def on_message(message, data) -> None:
self._reactor.schedule(lambda: self._on_message(message, data))
self._ui = ui
agent_source = (Path(__file__).parent / "itracer_agent.js").read_text(encoding="utf-8")
try:
tracer_script = session.create_script(name="itracer", source=agent_source, runtime=runtime)
self._tracer_script = tracer_script
self._ui._on_script_created(tracer_script)
tracer_script.on("message", on_message)
tracer_script.load()
tracer_api = tracer_script.exports_sync
outpath = ui.get_trace_output_path()
if outpath is None:
outpath = ui.prompt_for_trace_output_path(suggested_name=tracer_api.query_program_name() + ".itrace")
if outpath is None:
ui.on_trace_stopped("Missing output path")
return
self._outfile = open(outpath, "wb")
self._outfile.write(self.FILE_MAGIC)
strategy = ui.get_trace_strategy()
if strategy is None:
strategy = ui.prompt_for_trace_strategy(threads=tracer_api.list_threads())
if strategy is None:
ui.on_trace_stopped("Missing strategy")
return
buffer_location = tracer_api.create_buffer()
try:
system_session = device.attach(0)
reader_script = system_session.create_script(name="itracer", source=agent_source, runtime=runtime)
self._reader_script = reader_script
self._ui._on_script_created(reader_script)
reader_script.on("message", on_message)
reader_script.load()
reader_script.exports_sync.open_buffer(buffer_location)
except:
if self._reader_script is not None:
self._reader_script.unload()
self._reader_script = None
reader_script = None
if reader_script is not None:
reader_api = reader_script.exports_sync
else:
reader_api = tracer_script.exports_sync
self._reader_api = reader_api
reader_api.launch_buffer_reader()
tracer_script.exports_sync.launch_trace_session(strategy)
ui.on_trace_started()
except RPCException as e:
ui.on_trace_stopped(f"Unable to start: {e.args[0]}")
except Exception as e:
ui.on_trace_stopped(str(e))
except KeyboardInterrupt:
ui.on_trace_stopped()
def _on_message(self, message, data) -> None:
handled = False
if message["type"] == "send":
try:
payload = message["payload"]
mtype = payload["type"]
params = (mtype, payload, data)
except:
params = None
if params is not None:
handled = self._try_handle_message(*params)
if not handled:
print(message)
def _try_handle_message(self, mtype, message, data) -> bool:
if not mtype.startswith("itrace:"):
return False
if mtype == "itrace:chunk":
self._write_chunk(data)
else:
self._write_message(message, data)
if mtype == "itrace:compile":
self._total_blocks += 1
self._update_progress()
if mtype == "itrace:end":
self._ui.on_trace_stopped()
return True
def _update_progress(self) -> None:
self._ui.on_trace_progress(self._total_blocks, self._outfile.tell())
def _write_message(self, message, data) -> None:
f = self._outfile
raw_message = json.dumps(message).encode("utf-8")
f.write(struct.pack(">II", RecordType.MESSAGE, len(raw_message)))
f.write(raw_message)
data_size = len(data) if data is not None else 0
f.write(struct.pack(">I", data_size))
if data_size != 0:
f.write(data)
f.flush()
def _write_chunk(self, chunk) -> None:
f = self._outfile
f.write(struct.pack(">II", RecordType.CHUNK, len(chunk)))
f.write(chunk)
f.flush()
class RecordType:
MESSAGE = 1
CHUNK = 2
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,81 @@
import argparse
from typing import Any, List, MutableMapping
def main() -> None:
from frida_tools.application import ConsoleApplication, await_ctrl_c
class JoinApplication(ConsoleApplication):
def __init__(self) -> None:
ConsoleApplication.__init__(self, await_ctrl_c)
self._parsed_options: MutableMapping[str, Any] = {}
def _usage(self) -> str:
return "%(prog)s [options] target portal-location [portal-certificate] [portal-token]"
def _add_options(self, parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"--portal-location", help="join portal at LOCATION", metavar="LOCATION", dest="portal_location"
)
parser.add_argument(
"--portal-certificate",
help="speak TLS with portal, expecting CERTIFICATE",
metavar="CERTIFICATE",
dest="portal_certificate",
)
parser.add_argument(
"--portal-token", help="authenticate with portal using TOKEN", metavar="TOKEN", dest="portal_token"
)
parser.add_argument(
"--portal-acl-allow",
help="limit portal access to control channels with TAG",
metavar="TAG",
action="append",
dest="portal_acl",
)
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
location = args[0] if len(args) >= 1 else options.portal_location
certificate = args[1] if len(args) >= 2 else options.portal_certificate
token = args[2] if len(args) >= 3 else options.portal_token
acl = options.portal_acl
if location is None:
parser.error("portal location must be specified")
if certificate is not None:
self._parsed_options["certificate"] = certificate
if token is not None:
self._parsed_options["token"] = token
if acl is not None:
self._parsed_options["acl"] = acl
self._location = location
def _needs_target(self) -> bool:
return True
def _start(self) -> None:
self._update_status("Joining portal...")
try:
assert self._session is not None
self._session.join_portal(self._location, **self._parsed_options)
except Exception as e:
self._update_status("Unable to join: " + str(e))
self._exit(1)
return
self._update_status("Joined!")
self._exit(0)
def _stop(self) -> None:
pass
app = JoinApplication()
app.run()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

View File

@@ -0,0 +1,42 @@
import argparse
from typing import List
import frida
from frida_tools.application import ConsoleApplication, expand_target, infer_target
class KillApplication(ConsoleApplication):
def _usage(self) -> str:
return "%(prog)s [options] process"
def _add_options(self, parser: argparse.ArgumentParser) -> None:
parser.add_argument("process", help="process name or pid")
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
process = expand_target(infer_target(options.process))
if process[0] == "file":
parser.error("process name or pid must be specified")
self._process = process[1]
def _start(self) -> None:
try:
assert self._device is not None
self._device.kill(self._process)
except frida.ProcessNotFoundError:
self._update_status(f"unable to find process: {self._process}")
self._exit(1)
self._exit(0)
def main() -> None:
app = KillApplication()
app.run()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

View File

@@ -0,0 +1,139 @@
import argparse
import codecs
import os
from datetime import datetime, timezone
from operator import itemgetter
from typing import Any, List
from colorama import Fore, Style
from frida_tools.application import ConsoleApplication
STYLE_DIR = Fore.BLUE + Style.BRIGHT
STYLE_EXECUTABLE = Fore.GREEN + Style.BRIGHT
STYLE_LINK = Fore.CYAN + Style.BRIGHT
STYLE_ERROR = Fore.RED + Style.BRIGHT
def main() -> None:
app = LsApplication()
app.run()
class LsApplication(ConsoleApplication):
def _add_options(self, parser: argparse.ArgumentParser) -> None:
parser.add_argument("files", help="files to list information about", nargs="*")
def _usage(self) -> str:
return "%(prog)s [options] [FILE]..."
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
self._files = options.files
def _needs_target(self) -> bool:
return False
def _start(self) -> None:
try:
self._attach(0)
data_dir = os.path.dirname(__file__)
with codecs.open(os.path.join(data_dir, "fs_agent.js"), "r", "utf-8") as f:
source = f.read()
def on_message(message: Any, data: Any) -> None:
print(message)
assert self._session is not None
script = self._session.create_script(name="ls", source=source)
script.on("message", on_message)
self._on_script_created(script)
script.load()
groups = script.exports_sync.ls(self._files)
except Exception as e:
self._update_status(f"Failed to retrieve listing: {e}")
self._exit(1)
return
exit_status = 0
for i, group in enumerate(sorted(groups, key=lambda g: g["path"])):
path = group["path"]
if path != "" and len(groups) > 1:
if i > 0:
self._print("")
self._print(path + ":")
for path, message in group["errors"]:
self._print(STYLE_ERROR + message + Style.RESET_ALL)
exit_status = 2
rows = []
for name, target, type, access, nlink, owner, group, size, raw_mtime in group["entries"]:
mtime = datetime.fromtimestamp(raw_mtime / 1000.0, tz=timezone.utc)
rows.append((type + access, str(nlink), owner, group, str(size), mtime.strftime("%c"), name, target))
if len(rows) == 0:
break
widths = []
for column_index in range(len(rows[0]) - 2):
width = max(map(lambda row: len(row[column_index]), rows))
widths.append(width)
adjustments = [
"",
">",
"<",
"<",
">",
"<",
]
col_formats = []
for i, width in enumerate(widths):
adj = adjustments[i]
if adj != "":
fmt = "{:" + adj + str(width) + "}"
else:
fmt = "{}"
col_formats.append(fmt)
row_description = " ".join(col_formats)
for row in sorted(rows, key=itemgetter(6)):
meta_fields = row_description.format(*row[:-2])
name, target = row[6:8]
ftype_and_perms = row[0]
ftype = ftype_and_perms[0]
fperms = ftype_and_perms[1:]
name = format_name(name, ftype, fperms, target)
self._print(meta_fields + " " + name)
self._exit(exit_status)
def format_name(name: str, ftype: str, fperms: str, target) -> str:
if ftype == "l":
target_path, target_details = target
if target_details is not None:
target_type, target_perms = target_details
target_summary = format_name(target_path, target_type, target_perms, None)
else:
target_summary = STYLE_ERROR + target_path + Style.RESET_ALL
return STYLE_LINK + name + Style.RESET_ALL + " -> " + target_summary
if ftype == "d":
return STYLE_DIR + name + Style.RESET_ALL
if "x" in fperms:
return STYLE_EXECUTABLE + name + Style.RESET_ALL
return name
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

View File

@@ -0,0 +1,116 @@
def main() -> None:
import functools
import frida
from frida_tools.application import ConsoleApplication
class LSDApplication(ConsoleApplication):
def _usage(self) -> str:
return "%(prog)s [options]"
def _needs_device(self) -> bool:
return False
def _start(self) -> None:
try:
devices = frida.enumerate_devices()
except Exception as e:
self._update_status(f"Failed to enumerate devices: {e}")
self._exit(1)
return
device_name = {}
device_os = {}
for device in devices:
device_name[device.id] = device.name
try:
params = device.query_system_parameters()
except:
continue
device_name[device.id] = params.get("name", device.name)
os = params["os"]
version = os.get("version")
if version is not None:
device_os[device.id] = os["name"] + " " + version
else:
device_os[device.id] = os["name"]
id_column_width = max(map(lambda device: len(device.id) if device.id is not None else 0, devices))
type_column_width = max(map(lambda device: len(device.type) if device.type is not None else 0, devices))
name_column_width = max(map(lambda name: len(name) if name is not None else 0, device_name.values()))
os_column_width = max(map(lambda os: len(os) if os is not None else 0, device_os.values()))
header_format = (
"%-"
+ str(id_column_width)
+ "s "
+ "%-"
+ str(type_column_width)
+ "s "
+ "%-"
+ str(name_column_width)
+ "s "
+ "%-"
+ str(os_column_width)
+ "s"
)
self._print(header_format % ("Id", "Type", "Name", "OS"))
self._print(
f"{id_column_width * '-'} {type_column_width * '-'} {name_column_width * '-'} {os_column_width * '-'}"
)
line_format = (
"%-"
+ str(id_column_width)
+ "s "
+ "%-"
+ str(type_column_width)
+ "s "
+ "%-"
+ str(name_column_width)
+ "s "
+ "%-"
+ str(os_column_width)
+ "s"
)
for device in sorted(devices, key=functools.cmp_to_key(compare_devices)):
self._print(
line_format % (device.id, device.type, device_name.get(device.id), device_os.get(device.id, ""))
)
self._exit(0)
def compare_devices(a: frida.core.Device, b: frida.core.Device) -> int:
a_score = score(a)
b_score = score(b)
if a_score == b_score:
if a.name is None or b.name is None:
return 0
if a.name > b.name:
return 1
elif a.name < b.name:
return -1
else:
return 0
else:
if a_score > b_score:
return -1
elif a_score < b_score:
return 1
else:
return 0
def score(device: frida.core.Device) -> int:
type = device.type
if type == "local":
return 3
elif type == "usb":
return 2
else:
return 1
app = LSDApplication()
app.run()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

View File

@@ -0,0 +1,79 @@
class Module:
def __init__(self, name: str, base_address: int, size: int, path: str) -> None:
self.name = name
self.base_address = base_address
self.size = size
self.path = path
def __repr__(self) -> str:
return 'Module(name="%s", base_address=0x%x, size=%d, path="%s")' % (
self.name,
self.base_address,
self.size,
self.path,
)
def __hash__(self) -> int:
return self.base_address.__hash__()
def __eq__(self, other: object) -> bool:
return isinstance(other, Module) and self.base_address == other.base_address
def __ne__(self, other: object) -> bool:
return not (isinstance(other, Module) and self.base_address == other.base_address)
class Function:
def __init__(self, name: str, absolute_address: int) -> None:
self.name = name
self.absolute_address = absolute_address
def __str__(self) -> str:
return self.name
def __repr__(self) -> str:
return 'Function(name="%s", absolute_address=0x%x)' % (self.name, self.absolute_address)
def __hash__(self) -> int:
return self.absolute_address.__hash__()
def __eq__(self, other: object) -> bool:
return isinstance(other, Function) and self.absolute_address == other.absolute_address
def __ne__(self, other: object) -> bool:
return not (isinstance(other, Function) and self.absolute_address == other.absolute_address)
class ModuleFunction(Function):
def __init__(self, module: Module, name: str, relative_address: int, exported: bool) -> None:
super().__init__(name, module.base_address + relative_address)
self.module = module
self.relative_address = relative_address
self.exported = exported
def __repr__(self) -> str:
return 'ModuleFunction(module="%s", name="%s", relative_address=0x%x)' % (
self.module.name,
self.name,
self.relative_address,
)
class ObjCMethod(Function):
def __init__(self, mtype: str, cls: str, method: str, address: int) -> None:
self.mtype = mtype
self.cls = cls
self.method = method
self.address = address
super().__init__(self.display_name(), address)
def display_name(self) -> str:
return "{mtype}[{cls} {method}]".format(mtype=self.mtype, cls=self.cls, method=self.method)
def __repr__(self) -> str:
return 'ObjCMethod(mtype="%s", cls="%s", method="%s", address=0x%x)' % (
self.mtype,
self.cls,
self.method,
self.address,
)

View File

@@ -0,0 +1,286 @@
def main() -> None:
import argparse
import functools
import json
import math
import platform
import sys
from base64 import b64encode
from typing import List, Tuple, Union
try:
import termios
import tty
except:
pass
import _frida
from frida_tools.application import ConsoleApplication
class PSApplication(ConsoleApplication):
def _add_options(self, parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"-a",
"--applications",
help="list only applications",
action="store_true",
dest="list_only_applications",
default=False,
)
parser.add_argument(
"-i",
"--installed",
help="include all installed applications",
action="store_true",
dest="include_all_applications",
default=False,
)
parser.add_argument(
"-j",
"--json",
help="output results as JSON",
action="store_const",
dest="output_format",
const="json",
default="text",
)
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
if options.include_all_applications and not options.list_only_applications:
parser.error("-i cannot be used without -a")
self._list_only_applications = options.list_only_applications
self._include_all_applications = options.include_all_applications
self._output_format = options.output_format
self._terminal_type, self._icon_size = self._detect_terminal()
def _usage(self) -> str:
return "%(prog)s [options]"
def _start(self) -> None:
if self._list_only_applications:
self._list_applications()
else:
self._list_processes()
def _list_processes(self) -> None:
if self._output_format == "text" and self._terminal_type == "iterm2":
scope = "full"
else:
scope = "minimal"
try:
assert self._device is not None
processes = self._device.enumerate_processes(scope=scope)
except Exception as e:
self._update_status(f"Failed to enumerate processes: {e}")
self._exit(1)
return
if self._output_format == "text":
if len(processes) > 0:
pid_column_width = max(map(lambda p: len(str(p.pid)), processes))
icon_width = max(map(compute_icon_width, processes))
name_column_width = icon_width + max(map(lambda p: len(p.name), processes))
header_format = "%" + str(pid_column_width) + "s %s"
self._print(header_format % ("PID", "Name"))
self._print(f"{pid_column_width * '-'} {name_column_width * '-'}")
line_format = "%" + str(pid_column_width) + "d %s"
name_format = "%-" + str(name_column_width - icon_width) + "s"
for process in sorted(processes, key=functools.cmp_to_key(compare_processes)):
if icon_width != 0:
icons = process.parameters.get("icons", None)
if icons is not None:
icon = self._render_icon(icons[0])
else:
icon = " "
name = icon + " " + name_format % process.name
else:
name = name_format % process.name
self._print(line_format % (process.pid, name))
else:
self._log("error", "No running processes.")
elif self._output_format == "json":
result = []
for process in sorted(processes, key=functools.cmp_to_key(compare_processes)):
result.append({"pid": process.pid, "name": process.name})
self._print(json.dumps(result, sort_keys=False, indent=2))
self._exit(0)
def _list_applications(self) -> None:
if self._output_format == "text" and self._terminal_type == "iterm2":
scope = "full"
else:
scope = "minimal"
try:
assert self._device is not None
applications = self._device.enumerate_applications(scope=scope)
except Exception as e:
self._update_status(f"Failed to enumerate applications: {e}")
self._exit(1)
return
if not self._include_all_applications:
applications = list(filter(lambda app: app.pid != 0, applications))
if self._output_format == "text":
if len(applications) > 0:
pid_column_width = max(map(lambda app: len(str(app.pid)), applications))
icon_width = max(map(compute_icon_width, applications))
name_column_width = icon_width + max(map(lambda app: len(app.name), applications))
identifier_column_width = max(map(lambda app: len(app.identifier), applications))
header_format = (
"%"
+ str(pid_column_width)
+ "s "
+ "%-"
+ str(name_column_width)
+ "s "
+ "%-"
+ str(identifier_column_width)
+ "s"
)
self._print(header_format % ("PID", "Name", "Identifier"))
self._print(f"{pid_column_width * '-'} {name_column_width * '-'} {identifier_column_width * '-'}")
line_format = "%" + str(pid_column_width) + "s %s %-" + str(identifier_column_width) + "s"
name_format = "%-" + str(name_column_width - icon_width) + "s"
for app in sorted(applications, key=functools.cmp_to_key(compare_applications)):
if icon_width != 0:
icons = app.parameters.get("icons", None)
if icons is not None:
icon = self._render_icon(icons[0])
else:
icon = " "
name = icon + " " + name_format % app.name
else:
name = name_format % app.name
if app.pid == 0:
self._print(line_format % ("-", name, app.identifier))
else:
self._print(line_format % (app.pid, name, app.identifier))
elif self._include_all_applications:
self._log("error", "No installed applications.")
else:
self._log("error", "No running applications.")
elif self._output_format == "json":
result = []
if len(applications) > 0:
for app in sorted(applications, key=functools.cmp_to_key(compare_applications)):
result.append({"pid": (app.pid or None), "name": app.name, "identifier": app.identifier})
self._print(json.dumps(result, sort_keys=False, indent=2))
self._exit(0)
def _render_icon(self, icon) -> str:
return "\033]1337;File=inline=1;width={}px;height={}px;:{}\007".format(
self._icon_size, self._icon_size, b64encode(icon["image"]).decode("ascii")
)
def _detect_terminal(self) -> Tuple[str, int]:
icon_size = 0
if not self._have_terminal or self._plain_terminal or platform.system() != "Darwin":
return ("simple", icon_size)
fd = sys.stdin.fileno()
old_attributes = termios.tcgetattr(fd)
try:
tty.setraw(fd)
new_attributes = termios.tcgetattr(fd)
new_attributes[3] = new_attributes[3] & ~termios.ICANON & ~termios.ECHO
termios.tcsetattr(fd, termios.TCSANOW, new_attributes)
sys.stdout.write("\033[1337n")
sys.stdout.write("\033[5n")
sys.stdout.flush()
response = self._read_terminal_response("n")
if response not in ("0", "3"):
self._read_terminal_response("n")
if response.startswith("ITERM2 "):
version_tokens = response.split(" ", 1)[1].split(".", 2)
if len(version_tokens) >= 2 and int(version_tokens[0]) >= 3:
sys.stdout.write("\033[14t")
sys.stdout.flush()
height_in_pixels = int(self._read_terminal_response("t").split(";")[1])
sys.stdout.write("\033[18t")
sys.stdout.flush()
height_in_cells = int(self._read_terminal_response("t").split(";")[1])
icon_size = math.ceil((height_in_pixels / height_in_cells) * 1.77)
return ("iterm2", icon_size)
return ("simple", icon_size)
finally:
termios.tcsetattr(fd, termios.TCSANOW, old_attributes)
def _read_terminal_response(self, terminator: str) -> str:
sys.stdin.read(1)
sys.stdin.read(1)
result = ""
while True:
ch = sys.stdin.read(1)
if ch == terminator:
break
result += ch
return result
def compare_applications(a: _frida.Application, b: _frida.Application) -> int:
a_is_running = a.pid != 0
b_is_running = b.pid != 0
if a_is_running == b_is_running:
if a.name > b.name:
return 1
elif a.name < b.name:
return -1
else:
return 0
elif a_is_running:
return -1
else:
return 1
def compare_processes(a: _frida.Process, b: _frida.Process) -> int:
a_has_icon = "icons" in a.parameters
b_has_icon = "icons" in b.parameters
if a_has_icon == b_has_icon:
if a.name > b.name:
return 1
elif a.name < b.name:
return -1
else:
return 0
elif a_has_icon:
return -1
else:
return 1
def compute_icon_width(item: Union[_frida.Application, _frida.Process]) -> int:
for icon in item.parameters.get("icons", []):
if icon["format"] == "png":
return 4
return 0
app = PSApplication()
app.run()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

View File

@@ -0,0 +1,208 @@
import argparse
import codecs
import os
import sys
import time
import typing
from threading import Thread
from typing import Any, AnyStr, List, Mapping, Optional
import frida
from colorama import Fore, Style
from frida_tools.application import ConsoleApplication
from frida_tools.stream_controller import StreamController
from frida_tools.units import bytes_to_megabytes
def main() -> None:
app = PullApplication()
app.run()
class PullApplication(ConsoleApplication):
def _add_options(self, parser: argparse.ArgumentParser) -> None:
parser.add_argument("files", help="remote files to pull", nargs="+")
def _usage(self) -> str:
return "%(prog)s [options] REMOTE... LOCAL"
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
paths = options.files
if len(paths) == 1:
self._remote_paths = paths
self._local_paths = [os.path.join(os.getcwd(), basename_of_unknown_path(paths[0]))]
elif len(paths) == 2:
remote, local = paths
self._remote_paths = [remote]
if os.path.isdir(local):
self._local_paths = [os.path.join(local, basename_of_unknown_path(remote))]
else:
self._local_paths = [local]
else:
self._remote_paths = paths[:-1]
local_dir = paths[-1]
local_filenames = map(basename_of_unknown_path, self._remote_paths)
self._local_paths = [os.path.join(local_dir, filename) for filename in local_filenames]
self._script: Optional[frida.core.Script] = None
self._stream_controller: Optional[StreamController] = None
self._total_bytes = 0
self._time_started: Optional[float] = None
self._failed_paths = []
def _needs_target(self) -> bool:
return False
def _start(self) -> None:
try:
self._attach(0)
data_dir = os.path.dirname(__file__)
with codecs.open(os.path.join(data_dir, "fs_agent.js"), "r", "utf-8") as f:
source = f.read()
def on_message(message: Mapping[Any, Any], data: Any) -> None:
self._reactor.schedule(lambda: self._on_message(message, data))
assert self._session is not None
script = self._session.create_script(name="pull", source=source)
self._script = script
script.on("message", on_message)
self._on_script_created(script)
script.load()
self._stream_controller = StreamController(
self._post_stream_stanza,
self._on_incoming_stream_request,
on_stats_updated=self._on_stream_stats_updated,
)
worker = Thread(target=self._perform_pull)
worker.start()
except Exception as e:
self._update_status(f"Failed to pull: {e}")
self._exit(1)
return
def _stop(self) -> None:
if self._stream_controller is not None:
self._stream_controller.dispose()
def _perform_pull(self) -> None:
error = None
try:
assert self._script is not None
self._script.exports_sync.pull(self._remote_paths)
except Exception as e:
error = e
self._reactor.schedule(lambda: self._on_pull_finished(error))
def _on_pull_finished(self, error: Optional[Exception]) -> None:
for path, state in self._failed_paths:
if state == "partial":
try:
os.unlink(path)
except:
pass
if error is None:
self._render_summary_ui()
else:
self._print_error(str(error))
success = len(self._failed_paths) == 0 and error is None
status = 0 if success else 1
self._exit(status)
def _render_progress_ui(self) -> None:
assert self._stream_controller is not None
megabytes_received = bytes_to_megabytes(self._stream_controller.bytes_received)
total_megabytes = bytes_to_megabytes(self._total_bytes)
if total_megabytes != 0 and megabytes_received <= total_megabytes:
self._update_status(f"Pulled {megabytes_received:.1f} out of {total_megabytes:.1f} MB")
else:
self._update_status(f"Pulled {megabytes_received:.1f} MB")
def _render_summary_ui(self) -> None:
assert self._time_started is not None
duration = time.time() - self._time_started
if len(self._remote_paths) == 1:
prefix = f"{self._remote_paths[0]}: "
else:
prefix = ""
assert self._stream_controller is not None
sc = self._stream_controller
bytes_received = sc.bytes_received
megabytes_per_second = bytes_to_megabytes(bytes_received) / duration
self._update_status(
"{}{} file{} pulled. {:.1f} MB/s ({} bytes in {:.3f}s)".format(
prefix,
sc.streams_opened,
"s" if sc.streams_opened != 1 else "",
megabytes_per_second,
bytes_received,
duration,
)
)
def _on_message(self, message: Mapping[Any, Any], data: Any) -> None:
handled = False
if message["type"] == "send":
payload = message["payload"]
ptype = payload["type"]
if ptype == "stream":
stanza = payload["payload"]
assert self._stream_controller is not None
self._stream_controller.receive(stanza, data)
handled = True
elif ptype == "pull:status":
self._total_bytes = payload["total"]
self._time_started = time.time()
self._render_progress_ui()
handled = True
elif ptype == "pull:io-error":
index = payload["index"]
self._on_io_error(self._remote_paths[index], self._local_paths[index], payload["error"])
handled = True
if not handled:
self._print(message)
def _on_io_error(self, remote_path, local_path, error) -> None:
self._print_error(f"{remote_path}: {error}")
self._failed_paths.append((local_path, "partial"))
def _post_stream_stanza(self, stanza, data: Optional[AnyStr] = None) -> None:
self._script.post({"type": "stream", "payload": stanza}, data=data)
def _on_incoming_stream_request(self, label: str, details) -> typing.BinaryIO:
local_path = self._local_paths[int(label)]
try:
return open(local_path, "wb")
except Exception as e:
self._print_error(str(e))
self._failed_paths.append((local_path, "unopened"))
raise
def _on_stream_stats_updated(self) -> None:
self._render_progress_ui()
def _print_error(self, message: str) -> None:
self._print(Fore.RED + Style.BRIGHT + message + Style.RESET_ALL, file=sys.stderr)
def basename_of_unknown_path(path: str) -> str:
return path.replace("\\", "/").rsplit("/", 1)[-1]
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

View File

@@ -0,0 +1,208 @@
import argparse
import codecs
import os
import sys
import time
from threading import Event, Thread
from typing import AnyStr, List, MutableMapping, Optional
import frida
from colorama import Fore, Style
from frida_tools.application import ConsoleApplication
from frida_tools.stream_controller import DisposedException, StreamController
from frida_tools.units import bytes_to_megabytes
def main() -> None:
app = PushApplication()
app.run()
class PushApplication(ConsoleApplication):
def _add_options(self, parser: argparse.ArgumentParser) -> None:
parser.add_argument("files", help="local files to push", nargs="+")
def _usage(self) -> str:
return "%(prog)s [options] LOCAL... REMOTE"
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
paths = options.files
if len(paths) == 1:
raise ValueError("missing remote path")
self._local_paths = paths[:-1]
self._remote_path = paths[-1]
self._script: Optional[frida.core.Script] = None
self._stream_controller: Optional[StreamController] = None
self._total_bytes = 0
self._time_started: Optional[float] = None
self._completed = Event()
self._transfers: MutableMapping[str, bool] = {}
def _needs_target(self) -> bool:
return False
def _start(self) -> None:
try:
self._attach(0)
data_dir = os.path.dirname(__file__)
with codecs.open(os.path.join(data_dir, "fs_agent.js"), "r", "utf-8") as f:
source = f.read()
def on_message(message, data) -> None:
self._reactor.schedule(lambda: self._on_message(message, data))
assert self._session is not None
script = self._session.create_script(name="push", source=source)
self._script = script
script.on("message", on_message)
self._on_script_created(script)
script.load()
self._stream_controller = StreamController(
self._post_stream_stanza, on_stats_updated=self._on_stream_stats_updated
)
worker = Thread(target=self._perform_push)
worker.start()
except Exception as e:
self._update_status(f"Failed to push: {e}")
self._exit(1)
return
def _stop(self) -> None:
for path in self._local_paths:
if path not in self._transfers:
self._complete_transfer(path, success=False)
if self._stream_controller is not None:
self._stream_controller.dispose()
def _perform_push(self) -> None:
for path in self._local_paths:
try:
self._total_bytes += os.path.getsize(path)
except:
pass
self._time_started = time.time()
for i, path in enumerate(self._local_paths):
filename = os.path.basename(path)
try:
with open(path, "rb") as f:
assert self._stream_controller is not None
sink = self._stream_controller.open(str(i), {"filename": filename, "target": self._remote_path})
while True:
chunk = f.read(4 * 1024 * 1024)
if len(chunk) == 0:
break
sink.write(chunk)
sink.close()
except DisposedException:
break
except Exception as e:
self._print_error(str(e))
self._complete_transfer(path, success=False)
self._completed.wait()
self._reactor.schedule(lambda: self._on_push_finished())
def _on_push_finished(self) -> None:
successes = self._transfers.values()
if any(successes):
self._render_summary_ui()
status = 0 if all(successes) else 1
self._exit(status)
def _render_progress_ui(self) -> None:
if self._completed.is_set():
return
assert self._stream_controller is not None
megabytes_sent = bytes_to_megabytes(self._stream_controller.bytes_sent)
total_megabytes = bytes_to_megabytes(self._total_bytes)
if total_megabytes != 0 and megabytes_sent <= total_megabytes:
self._update_status(f"Pushed {megabytes_sent:.1f} out of {total_megabytes:.1f} MB")
else:
self._update_status(f"Pushed {megabytes_sent:.1f} MB")
def _render_summary_ui(self) -> None:
assert self._time_started is not None
duration = time.time() - self._time_started
if len(self._local_paths) == 1:
prefix = f"{self._local_paths[0]}: "
else:
prefix = ""
files_transferred = sum(map(int, self._transfers.values()))
assert self._stream_controller is not None
bytes_sent = self._stream_controller.bytes_sent
megabytes_per_second = bytes_to_megabytes(bytes_sent) / duration
self._update_status(
"{}{} file{} pushed. {:.1f} MB/s ({} bytes in {:.3f}s)".format(
prefix,
files_transferred,
"s" if files_transferred != 1 else "",
megabytes_per_second,
bytes_sent,
duration,
)
)
def _on_message(self, message, data) -> None:
handled = False
if message["type"] == "send":
payload = message["payload"]
ptype = payload["type"]
if ptype == "stream":
stanza = payload["payload"]
self._stream_controller.receive(stanza, data)
handled = True
elif ptype == "push:io-success":
index = payload["index"]
self._on_io_success(self._local_paths[index])
handled = True
elif ptype == "push:io-error":
index = payload["index"]
self._on_io_error(self._local_paths[index], payload["error"])
handled = True
if not handled:
self._print(message)
def _on_io_success(self, local_path: str) -> None:
self._complete_transfer(local_path, success=True)
def _on_io_error(self, local_path: str, error) -> None:
self._print_error(f"{local_path}: {error}")
self._complete_transfer(local_path, success=False)
def _complete_transfer(self, local_path: str, success: bool) -> None:
self._transfers[local_path] = success
if len(self._transfers) == len(self._local_paths):
self._completed.set()
def _post_stream_stanza(self, stanza, data: Optional[AnyStr] = None) -> None:
self._script.post({"type": "stream", "payload": stanza}, data=data)
def _on_stream_stats_updated(self) -> None:
self._render_progress_ui()
def _print_error(self, message: str) -> None:
self._print(Fore.RED + Style.BRIGHT + message + Style.RESET_ALL, file=sys.stderr)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

View File

@@ -0,0 +1,107 @@
import collections
import threading
import time
from typing import Callable, Deque, Optional, Tuple, Union
import frida
class Reactor:
"""
Run the given function until return in the main thread (or the thread of
the run method) and in a background thread receive and run additional tasks.
"""
def __init__(
self, run_until_return: Callable[["Reactor"], None], on_stop: Optional[Callable[[], None]] = None
) -> None:
self._running = False
self._run_until_return = run_until_return
self._on_stop = on_stop
self._pending: Deque[Tuple[Callable[[], None], Union[int, float]]] = collections.deque([])
self._lock = threading.Lock()
self._cond = threading.Condition(self._lock)
self.io_cancellable = frida.Cancellable()
self.ui_cancellable = frida.Cancellable()
self._ui_cancellable_fd = self.ui_cancellable.get_pollfd()
def __del__(self) -> None:
self._ui_cancellable_fd.release()
def is_running(self) -> bool:
with self._lock:
return self._running
def run(self) -> None:
with self._lock:
self._running = True
worker = threading.Thread(target=self._run)
worker.start()
self._run_until_return(self)
self.stop()
worker.join()
def _run(self) -> None:
running = True
while running:
now = time.time()
work = None
timeout = None
previous_pending_length = -1
with self._lock:
for item in self._pending:
(f, when) = item
if now >= when:
work = f
self._pending.remove(item)
break
if len(self._pending) > 0:
timeout = max([min(map(lambda item: item[1], self._pending)) - now, 0])
previous_pending_length = len(self._pending)
if work is not None:
with self.io_cancellable:
try:
work()
except frida.OperationCancelledError:
pass
with self._lock:
if self._running and len(self._pending) == previous_pending_length:
self._cond.wait(timeout)
running = self._running
if self._on_stop is not None:
self._on_stop()
self.ui_cancellable.cancel()
def stop(self) -> None:
self.schedule(self._stop)
def _stop(self) -> None:
with self._lock:
self._running = False
def schedule(self, f: Callable[[], None], delay: Optional[Union[int, float]] = None) -> None:
"""
append a function to the tasks queue of the reactor, optionally with a
delay in seconds
"""
now = time.time()
if delay is not None:
when = now + delay
else:
when = now
with self._lock:
self._pending.append((f, when))
self._cond.notify()
def cancel_io(self) -> None:
self.io_cancellable.cancel()

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,76 @@
import argparse
import codecs
import os
import sys
from typing import Any, List
from colorama import Fore, Style
from frida_tools.application import ConsoleApplication
def main() -> None:
app = RmApplication()
app.run()
class RmApplication(ConsoleApplication):
def _add_options(self, parser: argparse.ArgumentParser) -> None:
parser.add_argument("files", help="files to remove", nargs="+")
parser.add_argument("-f", "--force", help="ignore nonexistent files", action="store_true")
parser.add_argument(
"-r", "--recursive", help="remove directories and their contents recursively", action="store_true"
)
def _usage(self) -> str:
return "%(prog)s [options] FILE..."
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
self._paths = options.files
self._flags = []
if options.force:
self._flags.append("force")
if options.recursive:
self._flags.append("recursive")
def _needs_target(self) -> bool:
return False
def _start(self) -> None:
try:
self._attach(0)
data_dir = os.path.dirname(__file__)
with codecs.open(os.path.join(data_dir, "fs_agent.js"), "r", "utf-8") as f:
source = f.read()
def on_message(message: Any, data: Any) -> None:
self._reactor.schedule(lambda: self._on_message(message, data))
assert self._session is not None
script = self._session.create_script(name="pull", source=source)
script.on("message", on_message)
self._on_script_created(script)
script.load()
errors = script.exports_sync.rm(self._paths, self._flags)
for message in errors:
self._print(Fore.RED + Style.BRIGHT + message + Style.RESET_ALL, file=sys.stderr)
status = 0 if len(errors) == 0 else 1
self._exit(status)
except Exception as e:
self._update_status(str(e))
self._exit(1)
return
def _on_message(self, message: Any, data: Any) -> None:
print(message)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

View File

@@ -0,0 +1,182 @@
import threading
from typing import Any, AnyStr, BinaryIO, Callable, Mapping, Optional
class StreamController:
def __init__(
self,
post: Callable[[Any, Optional[AnyStr]], None],
on_incoming_stream_request: Optional[Callable[[Any, Any], BinaryIO]] = None,
on_incoming_stream_closed=None,
on_stats_updated=None,
) -> None:
self.streams_opened = 0
self.bytes_received = 0
self.bytes_sent = 0
self._handlers = {".create": self._on_create, ".finish": self._on_finish, ".write": self._on_write}
self._post = post
self._on_incoming_stream_request = on_incoming_stream_request
self._on_incoming_stream_closed = on_incoming_stream_closed
self._on_stats_updated = on_stats_updated
self._sources = {}
self._next_endpoint_id = 1
self._requests = {}
self._next_request_id = 1
def dispose(self) -> None:
error = DisposedException("disposed")
for request in self._requests.values():
request[2] = error
for event in [request[0] for request in self._requests.values()]:
event.set()
def open(self, label, details={}) -> "Sink":
eid = self._next_endpoint_id
self._next_endpoint_id += 1
endpoint = {"id": eid, "label": label, "details": details}
sink = Sink(self, endpoint)
self.streams_opened += 1
self._notify_stats_updated()
return sink
def receive(self, stanza: Mapping[str, Any], data: Any) -> None:
sid = stanza["id"]
name = stanza["name"]
payload = stanza.get("payload", None)
stype = name[0]
if stype == ".":
self._on_request(sid, name, payload, data)
elif stype == "+":
self._on_notification(sid, name, payload)
else:
raise ValueError("unknown stanza: " + name)
def _on_create(self, payload: Mapping[str, Any], data: Any) -> None:
endpoint = payload["endpoint"]
eid = endpoint["id"]
label = endpoint["label"]
details = endpoint["details"]
if self._on_incoming_stream_request is None:
raise ValueError("incoming streams not allowed")
source = self._on_incoming_stream_request(label, details)
self._sources[eid] = (source, label, details)
self.streams_opened += 1
self._notify_stats_updated()
def _on_finish(self, payload: Mapping[str, Any], data: Any) -> None:
eid = payload["endpoint"]["id"]
entry = self._sources.pop(eid, None)
if entry is None:
raise ValueError("invalid endpoint ID")
source, label, details = entry
source.close()
if self._on_incoming_stream_closed is not None:
self._on_incoming_stream_closed(label, details)
def _on_write(self, payload: Mapping[str, Any], data: Any) -> None:
entry = self._sources.get(payload["endpoint"]["id"], None)
if entry is None:
raise ValueError("invalid endpoint ID")
source, *_ = entry
source.write(data)
self.bytes_received += len(data)
self._notify_stats_updated()
def _request(self, name: str, payload: Mapping[Any, Any], data: Optional[AnyStr] = None):
rid = self._next_request_id
self._next_request_id += 1
completed = threading.Event()
request = [completed, None, None]
self._requests[rid] = request
self._post({"id": rid, "name": name, "payload": payload}, data)
completed.wait()
error = request[2]
if error is not None:
raise error
return request[1]
def _on_request(self, sid, name: str, payload: Mapping[str, Any], data: Any) -> None:
handler = self._handlers.get(name, None)
if handler is None:
raise ValueError("invalid request: " + name)
try:
result = handler(payload, data)
except Exception as e:
self._reject(sid, e)
return
self._resolve(sid, result)
def _resolve(self, sid, value) -> None:
self._post({"id": sid, "name": "+result", "payload": value})
def _reject(self, sid, error) -> None:
self._post({"id": sid, "name": "+error", "payload": {"message": str(error)}})
def _on_notification(self, sid, name: str, payload) -> None:
request = self._requests.pop(sid, None)
if request is None:
raise ValueError("invalid request ID")
if name == "+result":
request[1] = payload
elif name == "+error":
request[2] = StreamException(payload["message"])
else:
raise ValueError("unknown notification: " + name)
completed, *_ = request
completed.set()
def _notify_stats_updated(self) -> None:
if self._on_stats_updated is not None:
self._on_stats_updated()
class Sink:
def __init__(self, controller: StreamController, endpoint) -> None:
self._controller = controller
self._endpoint = endpoint
controller._request(".create", {"endpoint": endpoint})
def close(self) -> None:
self._controller._request(".finish", {"endpoint": self._endpoint})
def write(self, chunk) -> None:
ctrl = self._controller
ctrl._request(".write", {"endpoint": self._endpoint}, chunk)
ctrl.bytes_sent += len(chunk)
ctrl._notify_stats_updated()
class DisposedException(Exception):
pass
class StreamException(Exception):
pass

View File

@@ -0,0 +1,825 @@
import argparse
import binascii
import codecs
import os
import platform
import re
import subprocess
from typing import Callable, List, Optional, Union
import frida
from frida_tools.reactor import Reactor
def main() -> None:
import json
from colorama import Fore, Style
from frida_tools.application import ConsoleApplication, await_ctrl_c
class TracerApplication(ConsoleApplication, UI):
def __init__(self) -> None:
super().__init__(await_ctrl_c)
self._palette = [Fore.CYAN, Fore.MAGENTA, Fore.YELLOW, Fore.GREEN, Fore.RED, Fore.BLUE]
self._next_color = 0
self._attributes_by_thread_id = {}
self._last_event_tid = -1
def _add_options(self, parser: argparse.ArgumentParser) -> None:
pb = TracerProfileBuilder()
parser.add_argument(
"-I", "--include-module", help="include MODULE", metavar="MODULE", type=pb.include_modules
)
parser.add_argument(
"-X", "--exclude-module", help="exclude MODULE", metavar="MODULE", type=pb.exclude_modules
)
parser.add_argument(
"-i", "--include", help="include [MODULE!]FUNCTION", metavar="FUNCTION", type=pb.include
)
parser.add_argument(
"-x", "--exclude", help="exclude [MODULE!]FUNCTION", metavar="FUNCTION", type=pb.exclude
)
parser.add_argument(
"-a", "--add", help="add MODULE!OFFSET", metavar="MODULE!OFFSET", type=pb.include_relative_address
)
parser.add_argument("-T", "--include-imports", help="include program's imports", type=pb.include_imports)
parser.add_argument(
"-t",
"--include-module-imports",
help="include MODULE imports",
metavar="MODULE",
type=pb.include_imports,
)
parser.add_argument(
"-m",
"--include-objc-method",
help="include OBJC_METHOD",
metavar="OBJC_METHOD",
type=pb.include_objc_method,
)
parser.add_argument(
"-M",
"--exclude-objc-method",
help="exclude OBJC_METHOD",
metavar="OBJC_METHOD",
type=pb.exclude_objc_method,
)
parser.add_argument(
"-j",
"--include-java-method",
help="include JAVA_METHOD",
metavar="JAVA_METHOD",
type=pb.include_java_method,
)
parser.add_argument(
"-J",
"--exclude-java-method",
help="exclude JAVA_METHOD",
metavar="JAVA_METHOD",
type=pb.exclude_java_method,
)
parser.add_argument(
"-s",
"--include-debug-symbol",
help="include DEBUG_SYMBOL",
metavar="DEBUG_SYMBOL",
type=pb.include_debug_symbol,
)
parser.add_argument(
"-q", "--quiet", help="do not format output messages", action="store_true", default=False
)
parser.add_argument(
"-d",
"--decorate",
help="add module name to generated onEnter log statement",
action="store_true",
default=False,
)
parser.add_argument(
"-S",
"--init-session",
help="path to JavaScript file used to initialize the session",
metavar="PATH",
action="append",
default=[],
)
parser.add_argument(
"-P",
"--parameters",
help="parameters as JSON, exposed as a global named 'parameters'",
metavar="PARAMETERS_JSON",
)
parser.add_argument("-o", "--output", help="dump messages to file", metavar="OUTPUT")
self._profile_builder = pb
def _usage(self) -> str:
return "%(prog)s [options] target"
def _initialize(self, parser: argparse.ArgumentParser, options: argparse.Namespace, args: List[str]) -> None:
self._tracer: Optional[Tracer] = None
self._profile = self._profile_builder.build()
self._quiet: bool = options.quiet
self._decorate: bool = options.decorate
self._output: Optional[OutputFile] = None
self._output_path: str = options.output
self._init_scripts = []
for path in options.init_session:
with codecs.open(path, "rb", "utf-8") as f:
source = f.read()
self._init_scripts.append(InitScript(path, source))
if options.parameters is not None:
try:
params = json.loads(options.parameters)
except Exception as e:
raise ValueError(f"failed to parse parameters argument as JSON: {e}")
if not isinstance(params, dict):
raise ValueError("failed to parse parameters argument as JSON: not an object")
self._parameters = params
else:
self._parameters = {}
def _needs_target(self) -> bool:
return True
def _start(self) -> None:
if self._output_path is not None:
self._output = OutputFile(self._output_path)
stage = "early" if self._target[0] == "file" else "late"
self._tracer = Tracer(
self._reactor,
FileRepository(self._reactor, self._decorate),
self._profile,
self._init_scripts,
log_handler=self._log,
)
try:
self._tracer.start_trace(self._session, stage, self._parameters, self._runtime, self)
except Exception as e:
self._update_status(f"Failed to start tracing: {e}")
self._exit(1)
def _stop(self) -> None:
self._tracer.stop()
self._tracer = None
if self._output is not None:
self._output.close()
self._output = None
def on_script_created(self, script: frida.core.Script) -> None:
self._on_script_created(script)
def on_trace_progress(self, status: str, *params) -> None:
if status == "initializing":
self._update_status("Instrumenting...")
elif status == "initialized":
self._resume()
elif status == "started":
(count,) = params
if count == 1:
plural = ""
else:
plural = "s"
self._update_status("Started tracing %d function%s. Press Ctrl+C to stop." % (count, plural))
def on_trace_warning(self, message: str) -> None:
self._print(Fore.RED + Style.BRIGHT + "Warning" + Style.RESET_ALL + ": " + message)
def on_trace_error(self, message: str) -> None:
self._print(Fore.RED + Style.BRIGHT + "Error" + Style.RESET_ALL + ": " + message)
self._exit(1)
def on_trace_events(self, events) -> None:
no_attributes = Style.RESET_ALL
for timestamp, thread_id, depth, message in events:
if self._output is not None:
self._output.append(message + "\n")
elif self._quiet:
self._print(message)
else:
indent = depth * " | "
attributes = self._get_attributes(thread_id)
if thread_id != self._last_event_tid:
self._print("%s /* TID 0x%x */%s" % (attributes, thread_id, Style.RESET_ALL))
self._last_event_tid = thread_id
self._print("%6d ms %s%s%s%s" % (timestamp, attributes, indent, message, no_attributes))
def on_trace_handler_create(self, target, handler, source) -> None:
if self._quiet:
return
self._print('%s: Auto-generated handler at "%s"' % (target, source.replace("\\", "\\\\")))
def on_trace_handler_load(self, target, handler, source) -> None:
if self._quiet:
return
self._print('%s: Loaded handler at "%s"' % (target, source.replace("\\", "\\\\")))
def _get_attributes(self, thread_id):
attributes = self._attributes_by_thread_id.get(thread_id, None)
if attributes is None:
color = self._next_color
self._next_color += 1
attributes = self._palette[color % len(self._palette)]
if (1 + int(color / len(self._palette))) % 2 == 0:
attributes += Style.BRIGHT
self._attributes_by_thread_id[thread_id] = attributes
return attributes
app = TracerApplication()
app.run()
class TracerProfileBuilder:
def __init__(self) -> None:
self._spec = []
def include_modules(self, *module_name_globs: str) -> "TracerProfileBuilder":
for m in module_name_globs:
self._spec.append(("include", "module", m))
return self
def exclude_modules(self, *module_name_globs: str) -> "TracerProfileBuilder":
for m in module_name_globs:
self._spec.append(("exclude", "module", m))
return self
def include(self, *function_name_globs: str) -> "TracerProfileBuilder":
for f in function_name_globs:
self._spec.append(("include", "function", f))
return self
def exclude(self, *function_name_globs: str) -> "TracerProfileBuilder":
for f in function_name_globs:
self._spec.append(("exclude", "function", f))
return self
def include_relative_address(self, *address_rel_offsets: str) -> "TracerProfileBuilder":
for f in address_rel_offsets:
self._spec.append(("include", "relative-function", f))
return self
def include_imports(self, *module_name_globs: str) -> "TracerProfileBuilder":
for m in module_name_globs:
self._spec.append(("include", "imports", m))
return self
def include_objc_method(self, *function_name_globs: str) -> "TracerProfileBuilder":
for f in function_name_globs:
self._spec.append(("include", "objc-method", f))
return self
def exclude_objc_method(self, *function_name_globs: str) -> "TracerProfileBuilder":
for f in function_name_globs:
self._spec.append(("exclude", "objc-method", f))
return self
def include_java_method(self, *function_name_globs: str) -> "TracerProfileBuilder":
for f in function_name_globs:
self._spec.append(("include", "java-method", f))
return self
def exclude_java_method(self, *function_name_globs: str) -> "TracerProfileBuilder":
for f in function_name_globs:
self._spec.append(("exclude", "java-method", f))
return self
def include_debug_symbol(self, *function_name_globs: str) -> "TracerProfileBuilder":
for f in function_name_globs:
self._spec.append(("include", "debug-symbol", f))
return self
def build(self) -> "TracerProfile":
return TracerProfile(self._spec)
class TracerProfile:
def __init__(self, spec) -> None:
self.spec = spec
class Tracer:
def __init__(
self,
reactor: Reactor,
repository: "Repository",
profile: TracerProfile,
init_scripts=[],
log_handler: Callable[[str, str], None] = None,
) -> None:
self._reactor = reactor
self._repository = repository
self._profile = profile
self._script: Optional[frida.core.Script] = None
self._agent = None
self._init_scripts = init_scripts
self._log_handler = log_handler
def start_trace(self, session: frida.core.Session, stage, parameters, runtime, ui) -> None:
def on_create(*args) -> None:
ui.on_trace_handler_create(*args)
self._repository.on_create(on_create)
def on_load(*args) -> None:
ui.on_trace_handler_load(*args)
self._repository.on_load(on_load)
def on_update(target, handler, source) -> None:
self._agent.update(target.identifier, target.display_name, handler)
self._repository.on_update(on_update)
def on_message(message, data) -> None:
self._reactor.schedule(lambda: self._on_message(message, data, ui))
ui.on_trace_progress("initializing")
data_dir = os.path.dirname(__file__)
with codecs.open(os.path.join(data_dir, "tracer_agent.js"), "r", "utf-8") as f:
source = f.read()
script = session.create_script(name="tracer", source=source, runtime=runtime)
self._script = script
script.set_log_handler(self._log_handler)
script.on("message", on_message)
ui.on_script_created(script)
script.load()
self._agent = script.exports_sync
raw_init_scripts = [{"filename": script.filename, "source": script.source} for script in self._init_scripts]
self._agent.init(stage, parameters, raw_init_scripts, self._profile.spec)
def stop(self) -> None:
if self._script is not None:
try:
self._script.unload()
except:
pass
self._script = None
def _on_message(self, message, data, ui) -> None:
handled = False
if message["type"] == "send":
try:
payload = message["payload"]
mtype = payload["type"]
params = (mtype, payload, data, ui)
except:
# As user scripts may use send() we need to be prepared for this.
params = None
if params is not None:
handled = self._try_handle_message(*params)
if not handled:
print(message)
def _try_handle_message(self, mtype, params, data, ui) -> False:
if mtype == "events:add":
events = [
(timestamp, thread_id, depth, message)
for target_id, timestamp, thread_id, depth, message in params["events"]
]
ui.on_trace_events(events)
return True
if mtype == "handlers:get":
flavor = params["flavor"]
base_id = params["baseId"]
scripts = []
response = {"type": f"reply:{base_id}", "scripts": scripts}
repo = self._repository
next_id = base_id
for scope in params["scopes"]:
scope_name = scope["name"]
for member_name in scope["members"]:
target = TraceTarget(next_id, flavor, scope_name, member_name)
next_id += 1
handler = repo.ensure_handler(target)
scripts.append(handler)
self._script.post(response)
return True
if mtype == "agent:initialized":
ui.on_trace_progress("initialized")
return True
if mtype == "agent:started":
self._repository.commit_handlers()
ui.on_trace_progress("started", params["count"])
return True
if mtype == "agent:warning":
ui.on_trace_warning(params["message"])
return True
if mtype == "agent:error":
ui.on_trace_error(params["message"])
return True
return False
class TraceTarget:
def __init__(self, identifier, flavor, scope, name: Union[str, List[str]]) -> None:
self.identifier = identifier
self.flavor = flavor
self.scope = scope
if isinstance(name, list):
self.name = name[0]
self.display_name = name[1]
else:
self.name = name
self.display_name = name
def __str__(self) -> str:
return self.display_name
class Repository:
def __init__(self) -> None:
self._on_create_callback: Optional[Callable[[TraceTarget, str, str], None]] = None
self._on_load_callback: Optional[Callable[[TraceTarget, str, str], None]] = None
self._on_update_callback: Optional[Callable[[TraceTarget, str, str], None]] = None
self._decorate = False
def ensure_handler(self, target: TraceTarget):
raise NotImplementedError("not implemented")
def commit_handlers(self) -> None:
pass
def on_create(self, callback: Callable[[TraceTarget, str, str], None]) -> None:
self._on_create_callback = callback
def on_load(self, callback: Callable[[TraceTarget, str, str], None]) -> None:
self._on_load_callback = callback
def on_update(self, callback: Callable[[TraceTarget, str, str], None]) -> None:
self._on_update_callback = callback
def _notify_create(self, target: TraceTarget, handler: str, source: str) -> None:
if self._on_create_callback is not None:
self._on_create_callback(target, handler, source)
def _notify_load(self, target: TraceTarget, handler: str, source: str) -> None:
if self._on_load_callback is not None:
self._on_load_callback(target, handler, source)
def _notify_update(self, target: TraceTarget, handler: str, source: str) -> None:
if self._on_update_callback is not None:
self._on_update_callback(target, handler, source)
def _create_stub_handler(self, target: TraceTarget, decorate: bool) -> str:
if target.flavor == "java":
return self._create_stub_java_handler(target, decorate)
else:
return self._create_stub_native_handler(target, decorate)
def _create_stub_native_handler(self, target: TraceTarget, decorate: bool) -> str:
if target.flavor == "objc":
state = {"index": 2}
def objc_arg(m):
index = state["index"]
r = ":${args[%d]} " % index
state["index"] = index + 1
return r
log_str = "`" + re.sub(r":", objc_arg, target.display_name) + "`"
if log_str.endswith("} ]`"):
log_str = log_str[:-3] + "]`"
else:
for man_section in (2, 3):
args = []
try:
with open(os.devnull, "w") as devnull:
man_argv = ["man"]
if platform.system() != "Darwin":
man_argv.extend(["-E", "UTF-8"])
man_argv.extend(["-P", "col -b", str(man_section), target.name])
output = subprocess.check_output(man_argv, stderr=devnull)
match = re.search(
r"^SYNOPSIS(?:.|\n)*?((?:^.+$\n)* {5}\w+[ \*\n]*"
+ target.name
+ r"\((?:.+\,\s*?$\n)*?(?:.+\;$\n))(?:.|\n)*^DESCRIPTION",
output.decode("UTF-8", errors="replace"),
re.MULTILINE,
)
if match:
decl = match.group(1)
for argm in re.finditer(r"[\(,]\s*(.+?)\s*\b(\w+)(?=[,\)])", decl):
typ = argm.group(1)
arg = argm.group(2)
if arg == "void":
continue
if arg == "...":
args.append('", ..." +')
continue
read_ops = ""
annotate_pre = ""
annotate_post = ""
normalized_type = re.sub(r"\s+", "", typ)
if normalized_type.endswith("*restrict"):
normalized_type = normalized_type[:-8]
if normalized_type in ("char*", "constchar*"):
read_ops = ".readUtf8String()"
annotate_pre = '"'
annotate_post = '"'
arg_index = len(args)
args.append(
"%(arg_name)s=%(annotate_pre)s${args[%(arg_index)s]%(read_ops)s}%(annotate_post)s"
% {
"arg_name": arg,
"arg_index": arg_index,
"read_ops": read_ops,
"annotate_pre": annotate_pre,
"annotate_post": annotate_post,
}
)
break
except Exception:
pass
if decorate:
module_string = " [%s]" % os.path.basename(target.scope)
else:
module_string = ""
if len(args) == 0:
log_str = "'%(name)s()%(module_string)s'" % {"name": target.name, "module_string": module_string}
else:
log_str = "`%(name)s(%(args)s)%(module_string)s`" % {
"name": target.name,
"args": ", ".join(args),
"module_string": module_string,
}
return """\
/*
* Auto-generated by Frida. Please modify to match the signature of %(display_name)s.
* This stub is currently auto-generated from manpages when available.
*
* For full API reference, see: https://frida.re/docs/javascript-api/
*/
{
/**
* Called synchronously when about to call %(display_name)s.
*
* @this {object} - Object allowing you to store state for use in onLeave.
* @param {function} log - Call this function with a string to be presented to the user.
* @param {array} args - Function arguments represented as an array of NativePointer objects.
* For example use args[0].readUtf8String() if the first argument is a pointer to a C string encoded as UTF-8.
* It is also possible to modify arguments by assigning a NativePointer object to an element of this array.
* @param {object} state - Object allowing you to keep state across function calls.
* Only one JavaScript function will execute at a time, so do not worry about race-conditions.
* However, do not use this to store function arguments across onEnter/onLeave, but instead
* use "this" which is an object for keeping state local to an invocation.
*/
onEnter(log, args, state) {
log(%(log_str)s);
},
/**
* Called synchronously when about to return from %(display_name)s.
*
* See onEnter for details.
*
* @this {object} - Object allowing you to access state stored in onEnter.
* @param {function} log - Call this function with a string to be presented to the user.
* @param {NativePointer} retval - Return value represented as a NativePointer object.
* @param {object} state - Object allowing you to keep state across function calls.
*/
onLeave(log, retval, state) {
}
}
""" % {
"display_name": target.display_name,
"log_str": log_str,
}
def _create_stub_java_handler(self, target: TraceTarget, decorate) -> str:
return """\
/*
* Auto-generated by Frida. Please modify to match the signature of %(display_name)s.
*
* For full API reference, see: https://frida.re/docs/javascript-api/
*/
{
/**
* Called synchronously when about to call %(display_name)s.
*
* @this {object} - The Java class or instance.
* @param {function} log - Call this function with a string to be presented to the user.
* @param {array} args - Java method arguments.
* @param {object} state - Object allowing you to keep state across function calls.
*/
onEnter(log, args, state) {
log(`%(display_name)s(${args.map(JSON.stringify).join(', ')})`);
},
/**
* Called synchronously when about to return from %(display_name)s.
*
* See onEnter for details.
*
* @this {object} - The Java class or instance.
* @param {function} log - Call this function with a string to be presented to the user.
* @param {NativePointer} retval - Return value.
* @param {object} state - Object allowing you to keep state across function calls.
*/
onLeave(log, retval, state) {
if (retval !== undefined) {
log(`<= ${JSON.stringify(retval)}`);
}
}
}
""" % {
"display_name": target.display_name
}
class MemoryRepository(Repository):
def __init__(self) -> None:
super().__init__()
self._handlers = {}
def ensure_handler(self, target: TraceTarget) -> str:
handler = self._handlers.get(target)
if handler is None:
handler = self._create_stub_handler(target, False)
self._handlers[target] = handler
self._notify_create(target, handler, "memory")
else:
self._notify_load(target, handler, "memory")
return handler
class FileRepository(Repository):
def __init__(self, reactor: Reactor, decorate: bool) -> None:
super().__init__()
self._reactor = reactor
self._handler_by_id = {}
self._handler_by_file = {}
self._changed_files = set()
self._last_change_id = 0
self._repo_dir = os.path.join(os.getcwd(), "__handlers__")
self._repo_monitors = {}
self._decorate = decorate
def ensure_handler(self, target: TraceTarget) -> str:
entry = self._handler_by_id.get(target.identifier)
if entry is not None:
(target, handler, handler_file) = entry
return handler
handler = None
scope = target.scope
if len(scope) > 0:
handler_file = os.path.join(
self._repo_dir, to_filename(os.path.basename(scope)), to_handler_filename(target.name)
)
else:
handler_file = os.path.join(self._repo_dir, to_handler_filename(target.name))
if os.path.isfile(handler_file):
with codecs.open(handler_file, "r", "utf-8") as f:
handler = f.read()
self._notify_load(target, handler, handler_file)
if handler is None:
handler = self._create_stub_handler(target, self._decorate)
handler_dir = os.path.dirname(handler_file)
if not os.path.isdir(handler_dir):
os.makedirs(handler_dir)
with codecs.open(handler_file, "w", "utf-8") as f:
f.write(handler)
self._notify_create(target, handler, handler_file)
entry = (target, handler, handler_file)
self._handler_by_id[target.identifier] = entry
self._handler_by_file[handler_file] = entry
self._ensure_monitor(handler_file)
return handler
def _ensure_monitor(self, handler_file) -> None:
handler_dir = os.path.dirname(handler_file)
monitor = self._repo_monitors.get(handler_dir)
if monitor is None:
monitor = frida.FileMonitor(handler_dir)
monitor.on("change", self._on_change)
self._repo_monitors[handler_dir] = monitor
def commit_handlers(self) -> None:
for monitor in self._repo_monitors.values():
monitor.enable()
def _on_change(self, changed_file, other_file, event_type) -> None:
if changed_file not in self._handler_by_file or event_type == "changes-done-hint":
return
self._changed_files.add(changed_file)
self._last_change_id += 1
change_id = self._last_change_id
self._reactor.schedule(lambda: self._sync_handlers(change_id), delay=0.05)
def _sync_handlers(self, change_id) -> None:
if change_id != self._last_change_id:
return
changes = self._changed_files.copy()
self._changed_files.clear()
for changed_handler_file in changes:
(target, old_handler, handler_file) = self._handler_by_file[changed_handler_file]
with codecs.open(handler_file, "r", "utf-8") as f:
new_handler = f.read()
changed = new_handler != old_handler
if changed:
entry = (target, new_handler, handler_file)
self._handler_by_id[target.identifier] = entry
self._handler_by_file[handler_file] = entry
self._notify_update(target, new_handler, handler_file)
class InitScript:
def __init__(self, filename, source) -> None:
self.filename = filename
self.source = source
class OutputFile:
def __init__(self, filename: str) -> None:
self._fd = codecs.open(filename, "wb", "utf-8")
def close(self) -> None:
self._fd.close()
def append(self, message: str) -> None:
self._fd.write(message)
self._fd.flush()
class UI:
def on_script_created(self, script: frida.core.Script) -> None:
pass
def on_trace_progress(self, status) -> None:
pass
def on_trace_warning(self, message):
pass
def on_trace_error(self, message) -> None:
pass
def on_trace_events(self, events) -> None:
pass
def on_trace_handler_create(self, target, handler, source) -> None:
pass
def on_trace_handler_load(self, target, handler, source) -> None:
pass
def to_filename(name: str) -> str:
result = ""
for c in name:
if c.isalnum() or c == ".":
result += c
else:
result += "_"
return result
def to_handler_filename(name: str) -> str:
full_filename = to_filename(name)
if len(full_filename) <= 41:
return full_filename + ".js"
crc = binascii.crc32(full_filename.encode())
return full_filename[0:32] + "_%08x.js" % crc
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,2 @@
def bytes_to_megabytes(b: float) -> float:
return b / (1024 * 1024)