|
| 1 | +# -*- coding: utf-8 -*- |
| 2 | +# This file is part of Quark-Engine - https://github.com/quark-engine/quark-engine |
| 3 | +# See the file 'LICENSE' for copying permission. |
| 4 | + |
| 5 | +import functools |
| 6 | +import json |
| 7 | +import re |
| 8 | +import sys |
| 9 | +from dataclasses import dataclass |
| 10 | +from time import sleep |
| 11 | +from typing import Any, Dict, List, Tuple, Union |
| 12 | + |
| 13 | +import pkg_resources |
| 14 | +from quark.utils.regex import URL_REGEX |
| 15 | + |
| 16 | +import frida |
| 17 | +from frida.core import Device |
| 18 | +from frida.core import Session as FridaSession |
| 19 | + |
| 20 | +MethodCallEvent = Dict[str, Union[List[str], str]] |
| 21 | + |
| 22 | + |
| 23 | +class MethodCallEventDispatcher: |
| 24 | + def __init__(self, frida: FridaSession) -> None: |
| 25 | + self.frida = frida |
| 26 | + self.watchedMethods = {} |
| 27 | + |
| 28 | + @staticmethod |
| 29 | + def _getMethodIdentifier(targetMethod: str, paramType: str): |
| 30 | + return (targetMethod, paramType) |
| 31 | + |
| 32 | + def startWatchingMethodCall( |
| 33 | + self, targetMethod: str, methodParamTypes: str |
| 34 | + ) -> List[MethodCallEvent]: |
| 35 | + """Start tracking calls to the target method. |
| 36 | +
|
| 37 | + :param targetMethod: the target API |
| 38 | + :param methodParamTypes: the parameter types of the target API |
| 39 | + :return: python list that holds calls to the target method |
| 40 | + """ |
| 41 | + eventBuffer = [] |
| 42 | + methodId = self._getMethodIdentifier(targetMethod, methodParamTypes) |
| 43 | + |
| 44 | + self.watchedMethods[methodId] = eventBuffer |
| 45 | + self.script.exports.watch_method_call(targetMethod, methodParamTypes) |
| 46 | + |
| 47 | + return eventBuffer |
| 48 | + |
| 49 | + def stopWatchingMethodCall( |
| 50 | + self, targetMethod: str, methodParamTypes: str |
| 51 | + ) -> None: |
| 52 | + """Stop tracking calls to the target method. |
| 53 | +
|
| 54 | + :param targetMethod: the target API |
| 55 | + :param methodParamTypes: the parameter types of the target API |
| 56 | + """ |
| 57 | + methodId = self._getMethodIdentifier(targetMethod, methodParamTypes) |
| 58 | + |
| 59 | + if methodId in self.watchedMethods: |
| 60 | + del self.watchedMethods[methodId] |
| 61 | + |
| 62 | + def handleCapturedEvent(self, eventWrapperFromFrida: dict, _) -> None: |
| 63 | + """Send the event captured by Frida to the corresponding |
| 64 | + buffers. |
| 65 | +
|
| 66 | + :param eventWrapperFromFrida: python dict containing captured events |
| 67 | + """ |
| 68 | + if eventWrapperFromFrida["type"] == "error": |
| 69 | + errorDescription = eventWrapperFromFrida["description"] |
| 70 | + print(errorDescription, file=sys.stderr) |
| 71 | + return |
| 72 | + |
| 73 | + methodCallEvent = json.loads(eventWrapperFromFrida["payload"]) |
| 74 | + |
| 75 | + eventType = methodCallEvent.get("type", None) |
| 76 | + |
| 77 | + if eventType == "CallCaptured": |
| 78 | + methodId = tuple(methodCallEvent["identifier"][0:2]) |
| 79 | + |
| 80 | + if methodId in self.watchedMethods: |
| 81 | + messageBuffer = self.watchedMethods[methodId] |
| 82 | + messageBuffer.append(methodCallEvent) |
| 83 | + |
| 84 | + elif eventType == "FailedToWatch": |
| 85 | + methodId = tuple(methodCallEvent["identifier"]) |
| 86 | + self.watchedMethods.pop(methodId) |
| 87 | + |
| 88 | + |
| 89 | +@functools.lru_cache |
| 90 | +def _spawnApp( |
| 91 | + appPackageName: str, protocol="usb", **kwargs: Any |
| 92 | +) -> Tuple[Device, FridaSession, int]: |
| 93 | + """Spawn the target APP with Frida |
| 94 | +
|
| 95 | + :param appPackageName: the package name of the target APP |
| 96 | + :param protocol: string that holds the protocol to communicate with the |
| 97 | + Frida server, defaults to "usb" |
| 98 | + :return: tuple containing the device ID, the Frida instance and the process |
| 99 | + ID of the APP. |
| 100 | + """ |
| 101 | + device = None |
| 102 | + if protocol == "usb": |
| 103 | + device = frida.get_usb_device(**kwargs) |
| 104 | + elif protocol == "local": |
| 105 | + device = frida.get_local_device(**kwargs) |
| 106 | + elif protocol == "remote": |
| 107 | + device = frida.get_remote_device(**kwargs) |
| 108 | + |
| 109 | + processId = device.spawn([appPackageName]) |
| 110 | + session = device.attach(processId) |
| 111 | + |
| 112 | + return device, session, processId |
| 113 | + |
| 114 | + |
| 115 | +@functools.lru_cache |
| 116 | +def _injectAgent(frida: FridaSession) -> MethodCallEventDispatcher: |
| 117 | + """Inject a Frida agent to help track method calls. |
| 118 | +
|
| 119 | + :param frida: Frida instance to be injected |
| 120 | + :return: dispatcher that stores the captured calls to the appropriate |
| 121 | + buffers |
| 122 | + """ |
| 123 | + dispatcher = MethodCallEventDispatcher(frida) |
| 124 | + |
| 125 | + pathToFridaAgentSource = pkg_resources.resource_filename( |
| 126 | + "quark.script.frida", "agent.js" |
| 127 | + ) |
| 128 | + |
| 129 | + with open(pathToFridaAgentSource, "r") as fridaAgentSource: |
| 130 | + fridaAgent = dispatcher.frida.create_script(fridaAgentSource.read()) |
| 131 | + fridaAgent.on("message", dispatcher.handleCapturedEvent) |
| 132 | + fridaAgent.load() |
| 133 | + dispatcher.script = fridaAgent |
| 134 | + |
| 135 | + return dispatcher |
| 136 | + |
| 137 | + |
| 138 | +@dataclass |
| 139 | +class Behavior: |
| 140 | + _callEvent: MethodCallEvent |
| 141 | + |
| 142 | + def hasString(self, pattern: str, regex: bool = False) -> List[str]: |
| 143 | + """Check if the behavior contains strings |
| 144 | +
|
| 145 | + :param pattern: string to be checked |
| 146 | + :param regex: True if the string is a regular expression, defaults to |
| 147 | + False |
| 148 | + :return: python list containing all matched strings |
| 149 | + """ |
| 150 | + arguments = self.getParamValues() |
| 151 | + |
| 152 | + allMatchedStrings = set() |
| 153 | + for argument in arguments: |
| 154 | + if regex: |
| 155 | + matchedStrings = [ |
| 156 | + match.group(0) for match in re.finditer(pattern, argument) |
| 157 | + ] |
| 158 | + allMatchedStrings.update(matchedStrings) |
| 159 | + else: |
| 160 | + if pattern in argument: |
| 161 | + return [pattern] |
| 162 | + |
| 163 | + return list(allMatchedStrings) |
| 164 | + |
| 165 | + def hasUrl(self) -> List[str]: |
| 166 | + """Check if the behavior contains urls. |
| 167 | +
|
| 168 | + :return: python list containing all detected urls |
| 169 | + """ |
| 170 | + return self.hasString(URL_REGEX, True) |
| 171 | + |
| 172 | + def getParamValues(self) -> List[str]: |
| 173 | + """Get parameter values from behavior. |
| 174 | +
|
| 175 | + :return: python list containing parameter values |
| 176 | + """ |
| 177 | + return self._callEvent["paramValues"] |
| 178 | + |
| 179 | + |
| 180 | +@dataclass |
| 181 | +class FridaResult: |
| 182 | + _eventBuffer: List[MethodCallEvent] |
| 183 | + |
| 184 | + @property |
| 185 | + def behaviorOccurList(self) -> List[Behavior]: |
| 186 | + """List that stores instances of detected behavior in different part of |
| 187 | + the target file. |
| 188 | +
|
| 189 | + :return: detected behavior instance |
| 190 | + """ |
| 191 | + return [Behavior(message) for message in self._eventBuffer] |
| 192 | + |
| 193 | + |
| 194 | +def runFridaHook( |
| 195 | + apkPackageName: str, |
| 196 | + targetMethod: str, |
| 197 | + methodParamTypes: str, |
| 198 | + secondToWait: int = 10, |
| 199 | +) -> FridaResult: |
| 200 | + """Track calls to the specified method for given seconds. |
| 201 | +
|
| 202 | + :param apkPackageName: the package name of the target APP |
| 203 | + :param targetMethod: the target API |
| 204 | + :param methodParamTypes: string that holds the parameters used by the |
| 205 | + target API |
| 206 | + :param secondToWait: seconds to wait for method calls, defaults to 10 |
| 207 | + :return: FridaResult instance |
| 208 | + """ |
| 209 | + device, frida, appProcess = _spawnApp(apkPackageName) |
| 210 | + dispatcher = _injectAgent(frida) |
| 211 | + |
| 212 | + eventBuffer = dispatcher.startWatchingMethodCall( |
| 213 | + targetMethod, methodParamTypes |
| 214 | + ) |
| 215 | + device.resume(appProcess) |
| 216 | + |
| 217 | + sleep(secondToWait) |
| 218 | + dispatcher.stopWatchingMethodCall(targetMethod, methodParamTypes) |
| 219 | + |
| 220 | + return FridaResult(eventBuffer) |
0 commit comments