mirror of
https://github.com/mvt-project/mvt.git
synced 2026-02-16 02:12:46 +00:00
Compare commits
17 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1c912f68fe | ||
|
|
10a640d3f7 | ||
|
|
c3acc95e9e | ||
|
|
90d05336da | ||
|
|
5513e6e9e3 | ||
|
|
38116f8405 | ||
|
|
59b069f006 | ||
|
|
28e1348aa7 | ||
|
|
034338d1f4 | ||
|
|
09d5eabf2f | ||
|
|
a425d6c511 | ||
|
|
f8897a4f8c | ||
|
|
86eae68bdb | ||
|
|
d2bf348b03 | ||
|
|
25c6c03075 | ||
|
|
cf88740f6a | ||
|
|
eb4810b0ad |
@@ -199,6 +199,72 @@ def check_backup(ctx, iocs, output, backup_path, serial):
|
||||
run_module(m)
|
||||
|
||||
|
||||
#==============================================================================
|
||||
# Command: check-iocs
|
||||
#==============================================================================
|
||||
@cli.command("check-iocs", help="Compare stored JSON results to provided indicators")
|
||||
@click.option("--iocs", "-i", type=click.Path(exists=True), multiple=True,
|
||||
default=[], help=HELP_MSG_IOC)
|
||||
@click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
|
||||
@click.option("--module", "-m", help=HELP_MSG_MODULE)
|
||||
@click.argument("FOLDER", type=click.Path(exists=True))
|
||||
@click.pass_context
|
||||
def check_iocs(ctx, iocs, list_modules, module, folder):
|
||||
all_modules = []
|
||||
for entry in BACKUP_MODULES + ADB_MODULES:
|
||||
if entry not in all_modules:
|
||||
all_modules.append(entry)
|
||||
|
||||
if list_modules:
|
||||
log.info("Following is the list of available check-iocs modules:")
|
||||
for iocs_module in all_modules:
|
||||
log.info(" - %s", iocs_module.__name__)
|
||||
|
||||
return
|
||||
|
||||
log.info("Checking stored results against provided indicators...")
|
||||
|
||||
indicators = Indicators(log=log)
|
||||
indicators.load_indicators_files(iocs)
|
||||
|
||||
total_detections = 0
|
||||
for file_name in os.listdir(folder):
|
||||
name_only, ext = os.path.splitext(file_name)
|
||||
file_path = os.path.join(folder, file_name)
|
||||
|
||||
# TODO: Skipping processing of result files that are not json.
|
||||
# We might want to revisit this eventually.
|
||||
if ext != ".json":
|
||||
continue
|
||||
|
||||
for iocs_module in all_modules:
|
||||
if module and iocs_module.__name__ != module:
|
||||
continue
|
||||
|
||||
if iocs_module().get_slug() != name_only:
|
||||
continue
|
||||
|
||||
log.info("Loading results from \"%s\" with module %s", file_name,
|
||||
iocs_module.__name__)
|
||||
|
||||
m = iocs_module.from_json(file_path,
|
||||
log=logging.getLogger(iocs_module.__module__))
|
||||
if indicators.total_ioc_count > 0:
|
||||
m.indicators = indicators
|
||||
m.indicators.log = m.log
|
||||
|
||||
try:
|
||||
m.check_indicators()
|
||||
except NotImplementedError:
|
||||
continue
|
||||
else:
|
||||
total_detections += len(m.detected)
|
||||
|
||||
if total_detections > 0:
|
||||
log.warning("The check of the results produced %d detections!",
|
||||
total_detections)
|
||||
|
||||
|
||||
#==============================================================================
|
||||
# Command: download-iocs
|
||||
#==============================================================================
|
||||
|
||||
@@ -39,6 +39,10 @@ def get_virustotal_report(hashes):
|
||||
|
||||
|
||||
def virustotal_lookup(packages):
|
||||
# NOTE: This is temporary, until we resolved the issue.
|
||||
log.error("Unfortunately VirusTotal lookup is disabled until further notice, due to unresolved issues with the API service.")
|
||||
return
|
||||
|
||||
log.info("Looking up all extracted files on VirusTotal (www.virustotal.com)")
|
||||
|
||||
unique_hashes = []
|
||||
|
||||
@@ -11,14 +11,16 @@ from .dumpsys_packages import DumpsysPackages
|
||||
from .dumpsys_procstats import DumpsysProcstats
|
||||
from .dumpsys_receivers import DumpsysReceivers
|
||||
from .files import Files
|
||||
from .getprop import Getprop
|
||||
from .logcat import Logcat
|
||||
from .packages import Packages
|
||||
from .processes import Processes
|
||||
from .rootbinaries import RootBinaries
|
||||
from .root_binaries import RootBinaries
|
||||
from .settings import Settings
|
||||
from .sms import SMS
|
||||
from .whatsapp import Whatsapp
|
||||
|
||||
ADB_MODULES = [ChromeHistory, SMS, Whatsapp, Processes,
|
||||
ADB_MODULES = [ChromeHistory, SMS, Whatsapp, Processes, Getprop, Settings,
|
||||
DumpsysAccessibility, DumpsysBatterystats, DumpsysProcstats,
|
||||
DumpsysPackages, DumpsysReceivers, DumpsysFull,
|
||||
Packages, RootBinaries, Logcat, Files]
|
||||
|
||||
@@ -15,7 +15,7 @@ from adb_shell.adb_device import AdbDeviceTcp, AdbDeviceUsb
|
||||
from adb_shell.auth.keygen import keygen, write_public_keyfile
|
||||
from adb_shell.auth.sign_pythonrsa import PythonRSASigner
|
||||
from adb_shell.exceptions import (AdbCommandFailureException, DeviceAuthError,
|
||||
UsbReadFailedError)
|
||||
UsbDeviceNotFoundError, UsbReadFailedError)
|
||||
from usb1 import USBErrorAccess, USBErrorBusy
|
||||
|
||||
from mvt.common.module import InsufficientPrivileges, MVTModule
|
||||
@@ -65,7 +65,11 @@ class AndroidExtraction(MVTModule):
|
||||
# If no serial was specified or if the serial does not seem to be
|
||||
# a HOST:PORT definition, we use the USB transport.
|
||||
if not self.serial or ":" not in self.serial:
|
||||
self.device = AdbDeviceUsb(serial=self.serial)
|
||||
try:
|
||||
self.device = AdbDeviceUsb(serial=self.serial)
|
||||
except UsbDeviceNotFoundError:
|
||||
log.critical("No device found. Make sure it is connected and unlocked.")
|
||||
sys.exit(-1)
|
||||
# Otherwise we try to use the TCP transport.
|
||||
else:
|
||||
addr = self.serial.split(":")
|
||||
|
||||
@@ -9,10 +9,11 @@ from .base import AndroidExtraction
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
ACTION_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS"
|
||||
ACTION_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED"
|
||||
ACTION_DATA_SMS_RECEIVED = "android.intent.action.DATA_SMS_RECEIVED"
|
||||
ACTION_PHONE_STATE = "android.intent.action.PHONE_STATE"
|
||||
INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS"
|
||||
INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED"
|
||||
INTENT_DATA_SMS_RECEIVED = "android.intent.action.DATA_SMS_RECEIVED"
|
||||
INTENT_PHONE_STATE = "android.intent.action.PHONE_STATE"
|
||||
INTENT_NEW_OUTGOING_CALL = "android.intent.action.NEW_OUTGOING_CALL"
|
||||
|
||||
|
||||
class DumpsysReceivers(AndroidExtraction):
|
||||
@@ -24,6 +25,24 @@ class DumpsysReceivers(AndroidExtraction):
|
||||
output_folder=output_folder, fast_mode=fast_mode,
|
||||
log=log, results=results)
|
||||
|
||||
def check_indicators(self):
|
||||
for result in self.results:
|
||||
if result["activity"] == INTENT_NEW_OUTGOING_SMS:
|
||||
self.log.info("Found a receiver to intercept outgoing SMS messages: \"%s\"",
|
||||
result["receiver"])
|
||||
elif result["activity"] == INTENT_SMS_RECEIVED:
|
||||
self.log.info("Found a receiver to intercept incoming SMS messages: \"%s\"",
|
||||
result["receiver"])
|
||||
elif result["activity"] == INTENT_DATA_SMS_RECEIVED:
|
||||
self.log.info("Found a receiver to intercept incoming data SMS message: \"%s\"",
|
||||
result["receiver"])
|
||||
elif result["activity"] == INTENT_PHONE_STATE:
|
||||
self.log.info("Found a receiver monitoring telephony state/incoming calls: \"%s\"",
|
||||
result["receiver"])
|
||||
elif result["activity"] == INTENT_NEW_OUTGOING_CALL:
|
||||
self.log.info("Found a receiver monitoring outgoing calls: \"%s\"",
|
||||
result["receiver"])
|
||||
|
||||
def run(self):
|
||||
self._adb_connect()
|
||||
|
||||
@@ -34,17 +53,20 @@ class DumpsysReceivers(AndroidExtraction):
|
||||
activity = None
|
||||
for line in output.split("\n"):
|
||||
# Find activity block markers.
|
||||
if line.strip().startswith(ACTION_NEW_OUTGOING_SMS):
|
||||
activity = ACTION_NEW_OUTGOING_SMS
|
||||
if line.strip().startswith(INTENT_NEW_OUTGOING_SMS):
|
||||
activity = INTENT_NEW_OUTGOING_SMS
|
||||
continue
|
||||
elif line.strip().startswith(ACTION_SMS_RECEIVED):
|
||||
activity = ACTION_SMS_RECEIVED
|
||||
elif line.strip().startswith(INTENT_SMS_RECEIVED):
|
||||
activity = INTENT_SMS_RECEIVED
|
||||
continue
|
||||
elif line.strip().startswith(ACTION_PHONE_STATE):
|
||||
activity = ACTION_PHONE_STATE
|
||||
elif line.strip().startswith(INTENT_PHONE_STATE):
|
||||
activity = INTENT_PHONE_STATE
|
||||
continue
|
||||
elif line.strip().startswith(ACTION_DATA_SMS_RECEIVED):
|
||||
activity = ACTION_DATA_SMS_RECEIVED
|
||||
elif line.strip().startswith(INTENT_DATA_SMS_RECEIVED):
|
||||
activity = INTENT_DATA_SMS_RECEIVED
|
||||
continue
|
||||
elif line.strip().startswith(INTENT_NEW_OUTGOING_CALL):
|
||||
activity = INTENT_NEW_OUTGOING_CALL
|
||||
continue
|
||||
|
||||
# If we are not in an activity block yet, skip.
|
||||
@@ -65,19 +87,6 @@ class DumpsysReceivers(AndroidExtraction):
|
||||
if package_name == "com.google.android.gms":
|
||||
continue
|
||||
|
||||
if activity == ACTION_NEW_OUTGOING_SMS:
|
||||
self.log.info("Found a receiver to intercept outgoing SMS messages: \"%s\"",
|
||||
receiver)
|
||||
elif activity == ACTION_SMS_RECEIVED:
|
||||
self.log.info("Found a receiver to intercept incoming SMS messages: \"%s\"",
|
||||
receiver)
|
||||
elif activity == ACTION_DATA_SMS_RECEIVED:
|
||||
self.log.info("Found a receiver to intercept incoming data SMS message: \"%s\"",
|
||||
receiver)
|
||||
elif activity == ACTION_PHONE_STATE:
|
||||
self.log.info("Found a receiver monitoring telephony state: \"%s\"",
|
||||
receiver)
|
||||
|
||||
self.results.append({
|
||||
"activity": activity,
|
||||
"package_name": package_name,
|
||||
|
||||
@@ -22,30 +22,16 @@ class Files(AndroidExtraction):
|
||||
super().__init__(file_path=file_path, base_folder=base_folder,
|
||||
output_folder=output_folder, fast_mode=fast_mode,
|
||||
log=log, results=results)
|
||||
self.full_find = None
|
||||
self.full_find = False
|
||||
|
||||
def find_path(self, file_path):
|
||||
"""Checks if Android system supports full find command output"""
|
||||
# Check find command params on first run
|
||||
# Run find command with correct args and parse results.
|
||||
|
||||
# Check that full file printf options are suppported on first run.
|
||||
if self.full_find is None:
|
||||
output = self._adb_command("find '/' -maxdepth 1 -printf '%T@ %m %s %u %g %p\n' 2> /dev/null")
|
||||
if not (output or output.strip().splitlines()):
|
||||
# Full find command failed to generate output, fallback to basic file arguments
|
||||
self.full_find = False
|
||||
else:
|
||||
self.full_find = True
|
||||
|
||||
found_files = []
|
||||
if self.full_find is True:
|
||||
# Run full file command and collect additonal file information.
|
||||
def find_files(self, file_path):
|
||||
if self.full_find:
|
||||
output = self._adb_command(f"find '{file_path}' -printf '%T@ %m %s %u %g %p\n' 2> /dev/null")
|
||||
|
||||
for file_line in output.splitlines():
|
||||
[unix_timestamp, mode, size, owner, group, full_path] = file_line.rstrip().split(" ", 5)
|
||||
mod_time = convert_timestamp_to_iso(datetime.datetime.utcfromtimestamp(int(float(unix_timestamp))))
|
||||
found_files.append({
|
||||
self.results.append({
|
||||
"path": full_path,
|
||||
"modified_time": mod_time,
|
||||
"mode": mode,
|
||||
@@ -56,14 +42,9 @@ class Files(AndroidExtraction):
|
||||
"group": group,
|
||||
})
|
||||
else:
|
||||
# Run a basic listing of file paths.
|
||||
output = self._adb_command(f"find '{file_path}' 2> /dev/null")
|
||||
for file_line in output.splitlines():
|
||||
found_files.append({
|
||||
"path": file_line.rstrip()
|
||||
})
|
||||
|
||||
return found_files
|
||||
self.results.append({"path": file_line.rstrip()})
|
||||
|
||||
def serialize(self, record):
|
||||
if "modified_time" in record:
|
||||
@@ -85,6 +66,7 @@ class Files(AndroidExtraction):
|
||||
def check_indicators(self):
|
||||
"""Check file list for known suspicious files or suspicious properties"""
|
||||
self.check_suspicious()
|
||||
|
||||
if not self.indicators:
|
||||
return
|
||||
|
||||
@@ -95,25 +77,21 @@ class Files(AndroidExtraction):
|
||||
|
||||
def run(self):
|
||||
self._adb_connect()
|
||||
found_file_paths = []
|
||||
|
||||
DATA_PATHS = ["/data/local/tmp/", "/sdcard/", "/tmp/"]
|
||||
for path in DATA_PATHS:
|
||||
file_info = self.find_path(path)
|
||||
found_file_paths.extend(file_info)
|
||||
output = self._adb_command("find '/' -maxdepth 1 -printf '%T@ %m %s %u %g %p\n' 2> /dev/null")
|
||||
if output or output.strip().splitlines():
|
||||
self.full_find = True
|
||||
|
||||
# Store results
|
||||
self.results.extend(found_file_paths)
|
||||
self.log.info("Found %s files in primary Android data directories.", len(found_file_paths))
|
||||
for data_path in ["/data/local/tmp/", "/sdcard/", "/tmp/"]:
|
||||
self.find_files(data_path)
|
||||
|
||||
self.log.info("Found %s files in primary Android data directories", len(self.results))
|
||||
|
||||
if self.fast_mode:
|
||||
self.log.info("Flag --fast was enabled: skipping full file listing")
|
||||
else:
|
||||
self.log.info("Flag --fast was not enabled: processing full file listing. "
|
||||
"This may take a while...")
|
||||
output = self.find_path("/")
|
||||
if output and self.output_folder:
|
||||
self.results.extend(output)
|
||||
log.info("List of visible files stored in files.json")
|
||||
self.log.info("Processing full file listing. This may take a while...")
|
||||
self.find_files("/")
|
||||
self.log.info("Found %s total files", len(self.results))
|
||||
|
||||
self._adb_disconnect()
|
||||
|
||||
46
mvt/android/modules/adb/getprop.py
Normal file
46
mvt/android/modules/adb/getprop.py
Normal file
@@ -0,0 +1,46 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021 The MVT Project Authors.
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
|
||||
from .base import AndroidExtraction
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Getprop(AndroidExtraction):
|
||||
"""This module extracts device properties from getprop command."""
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
serial=None, fast_mode=False, log=None, results=[]):
|
||||
super().__init__(file_path=file_path, base_folder=base_folder,
|
||||
output_folder=output_folder, fast_mode=fast_mode,
|
||||
log=log, results=results)
|
||||
|
||||
self.results = {} if not results else results
|
||||
|
||||
def run(self):
|
||||
self._adb_connect()
|
||||
|
||||
rxp = re.compile("\\[(.+?)\\]: \\[(.+?)\\]")
|
||||
out = self._adb_command("getprop")
|
||||
for line in out.splitlines():
|
||||
line = line.strip()
|
||||
if line == "":
|
||||
continue
|
||||
|
||||
matches = re.findall(rxp, line)
|
||||
if not matches or len(matches[0]) != 2:
|
||||
continue
|
||||
|
||||
key = matches[0][0]
|
||||
value = matches[0][1]
|
||||
self.results[key] = value
|
||||
|
||||
self._adb_disconnect()
|
||||
|
||||
self.log.info("Extracted %d Android system properties", len(self.results))
|
||||
@@ -8,6 +8,9 @@ import os
|
||||
|
||||
import pkg_resources
|
||||
|
||||
from mvt.android.lookups.koodous import koodous_lookup
|
||||
from mvt.android.lookups.virustotal import virustotal_lookup
|
||||
|
||||
from .base import AndroidExtraction
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
@@ -42,9 +45,6 @@ class Packages(AndroidExtraction):
|
||||
return records
|
||||
|
||||
def check_indicators(self):
|
||||
if not self.indicators:
|
||||
return
|
||||
|
||||
root_packages_path = os.path.join("..", "..", "data", "root_packages.txt")
|
||||
root_packages_string = pkg_resources.resource_string(__name__, root_packages_path)
|
||||
root_packages = root_packages_string.decode("utf-8").split("\n")
|
||||
@@ -55,16 +55,19 @@ class Packages(AndroidExtraction):
|
||||
self.log.warning("Found an installed package related to rooting/jailbreaking: \"%s\"",
|
||||
result["package_name"])
|
||||
self.detected.append(result)
|
||||
if result["package_name"] in self.indicators.ioc_app_ids:
|
||||
self.log.warning("Found a malicious package name: \"%s\"",
|
||||
result["package_name"])
|
||||
continue
|
||||
|
||||
ioc = self.indicators.check_app_id(result.get("package_name"))
|
||||
if ioc:
|
||||
result["matched_indicators"] = ioc
|
||||
self.detected.append(result)
|
||||
for file in result["files"]:
|
||||
if file["sha256"] in self.indicators.ioc_files_sha256:
|
||||
self.log.warning("Found a malicious APK: \"%s\" %s",
|
||||
result["package_name"],
|
||||
file["sha256"])
|
||||
self.detected.append(result)
|
||||
continue
|
||||
|
||||
for package_file in result["files"]:
|
||||
ioc = self.indicators.check_file_hash(package_file["sha256"])
|
||||
if ioc:
|
||||
result["matched_indicators"] = ioc
|
||||
self.detected.append(result)
|
||||
|
||||
def _get_files_for_package(self, package_name):
|
||||
output = self._adb_command(f"pm path {package_name}")
|
||||
@@ -157,13 +160,19 @@ class Packages(AndroidExtraction):
|
||||
if result["package_name"] == package_name:
|
||||
self.results[i][cmd["field"]] = True
|
||||
|
||||
packages_to_lookup = []
|
||||
for result in self.results:
|
||||
if result["system"]:
|
||||
continue
|
||||
|
||||
packages_to_lookup.append(result)
|
||||
self.log.info("Found non-system package with name \"%s\" installed by \"%s\" on %s",
|
||||
result["package_name"], result["installer"], result["timestamp"])
|
||||
|
||||
if not self.fast_mode:
|
||||
virustotal_lookup(packages_to_lookup)
|
||||
koodous_lookup(packages_to_lookup)
|
||||
|
||||
self.log.info("Extracted at total of %d installed package names",
|
||||
len(self.results))
|
||||
|
||||
|
||||
106
mvt/android/modules/adb/settings.py
Normal file
106
mvt/android/modules/adb/settings.py
Normal file
@@ -0,0 +1,106 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021 The MVT Project Authors.
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
|
||||
from .base import AndroidExtraction
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
ANDROID_DANGEROUS_SETTINGS = [
|
||||
{
|
||||
"description": "disabled Google Play Services apps verification",
|
||||
"key": "verifier_verify_adb_installs",
|
||||
"safe_value": "1",
|
||||
},
|
||||
{
|
||||
"description": "disabled Google Play Protect",
|
||||
"key": "package_verifier_enable",
|
||||
"safe_value": "1",
|
||||
},
|
||||
{
|
||||
"description": "disabled Google Play Protect",
|
||||
"key": "package_verifier_user_consent",
|
||||
"safe_value": "1",
|
||||
},
|
||||
{
|
||||
"description": "disabled Google Play Protect",
|
||||
"key": "upload_apk_enable",
|
||||
"safe_value": "1",
|
||||
},
|
||||
{
|
||||
"description": "enabled installation of non-market apps",
|
||||
"key": "install_non_market_apps",
|
||||
"safe_value": "0",
|
||||
},
|
||||
{
|
||||
"description": "disabled confirmation of adb apps installation",
|
||||
"key": "adb_install_need_confirm",
|
||||
"safe_value": "1",
|
||||
},
|
||||
{
|
||||
"description": "disabled sharing of security reports",
|
||||
"key": "send_security_reports",
|
||||
"safe_value": "1",
|
||||
},
|
||||
{
|
||||
"description": "disabled sharing of crash logs with manufacturer",
|
||||
"key": "samsung_errorlog_agree",
|
||||
"safe_value": "1",
|
||||
},
|
||||
{
|
||||
"description": "disabled applications errors reports",
|
||||
"key": "send_action_app_error",
|
||||
"safe_value": "1",
|
||||
},
|
||||
]
|
||||
|
||||
class Settings(AndroidExtraction):
|
||||
"""This module extracts Android system settings."""
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
serial=None, fast_mode=False, log=None, results=[]):
|
||||
super().__init__(file_path=file_path, base_folder=base_folder,
|
||||
output_folder=output_folder, fast_mode=fast_mode,
|
||||
log=log, results=results)
|
||||
|
||||
self.results = {} if not results else results
|
||||
|
||||
def check_indicators(self):
|
||||
for namespace, settings in self.results.items():
|
||||
for key, value in settings.items():
|
||||
for danger in ANDROID_DANGEROUS_SETTINGS:
|
||||
# Check if one of the dangerous settings is using an unsafe
|
||||
# value (different than the one specified).
|
||||
if danger["key"] == key and danger["safe_value"] != value:
|
||||
self.log.warning("Found suspicious setting \"%s = %s\" (%s)",
|
||||
key, value, danger["description"])
|
||||
break
|
||||
|
||||
def run(self):
|
||||
self._adb_connect()
|
||||
|
||||
for namespace in ["system", "secure", "global"]:
|
||||
out = self._adb_command(f"cmd settings list {namespace}")
|
||||
if not out:
|
||||
continue
|
||||
|
||||
self.results[namespace] = {}
|
||||
|
||||
for line in out.splitlines():
|
||||
line = line.strip()
|
||||
if line == "":
|
||||
continue
|
||||
|
||||
fields = line.split("=", 1)
|
||||
try:
|
||||
self.results[namespace][fields[0]] = fields[1]
|
||||
except IndexError:
|
||||
continue
|
||||
|
||||
self._adb_disconnect()
|
||||
@@ -173,8 +173,7 @@ class Indicators:
|
||||
|
||||
:param url: URL to match against domain indicators
|
||||
:type url: str
|
||||
:returns: True if the URL matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
:returns: Indicator details if matched, otherwise None
|
||||
|
||||
"""
|
||||
# TODO: If the IOC domain contains a subdomain, it is not currently
|
||||
@@ -246,8 +245,7 @@ class Indicators:
|
||||
|
||||
:param urls: List of URLs to check against domain indicators
|
||||
:type urls: list
|
||||
:returns: True if any URL matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
:returns: Indicator details if matched, otherwise None
|
||||
|
||||
"""
|
||||
if not urls:
|
||||
@@ -264,8 +262,7 @@ class Indicators:
|
||||
|
||||
:param process: Process name to check against process indicators
|
||||
:type process: str
|
||||
:returns: True if process matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
:returns: Indicator details if matched, otherwise None
|
||||
|
||||
"""
|
||||
if not process:
|
||||
@@ -290,8 +287,7 @@ class Indicators:
|
||||
|
||||
:param processes: List of processes to check against process indicators
|
||||
:type processes: list
|
||||
:returns: True if process matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
:returns: Indicator details if matched, otherwise None
|
||||
|
||||
"""
|
||||
if not processes:
|
||||
@@ -307,8 +303,7 @@ class Indicators:
|
||||
|
||||
:param email: Email address to check against email indicators
|
||||
:type email: str
|
||||
:returns: True if email address matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
:returns: Indicator details if matched, otherwise None
|
||||
|
||||
"""
|
||||
if not email:
|
||||
@@ -326,8 +321,7 @@ class Indicators:
|
||||
:param file_name: File name to check against file
|
||||
indicators
|
||||
:type file_name: str
|
||||
:returns: True if the file name matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
:returns: Indicator details if matched, otherwise None
|
||||
|
||||
"""
|
||||
if not file_name:
|
||||
@@ -345,8 +339,7 @@ class Indicators:
|
||||
:param file_path: File path or file name to check against file
|
||||
indicators
|
||||
:type file_path: str
|
||||
:returns: True if the file path matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
:returns: Indicator details if matched, otherwise None
|
||||
|
||||
"""
|
||||
if not file_path:
|
||||
@@ -368,16 +361,53 @@ class Indicators:
|
||||
|
||||
:param profile_uuid: Profile UUID to check against configuration profile indicators
|
||||
:type profile_uuid: str
|
||||
:returns: True if the UUID in indicator list, otherwise False
|
||||
:rtype: bool
|
||||
:returns: Indicator details if matched, otherwise None
|
||||
|
||||
"""
|
||||
if not profile_uuid:
|
||||
return None
|
||||
|
||||
for ioc in self.get_iocs("ios_profile_ids"):
|
||||
if profile_uuid in ioc["value"]:
|
||||
self.log.warning("Found a known suspicious profile ID \"%s\" matching indicators from \"%s\"",
|
||||
profile_uuid, ioc["name"])
|
||||
return ioc
|
||||
|
||||
def check_file_hash(self, file_hash):
|
||||
"""Check the provided SHA256 file hash against the list of indicators.
|
||||
|
||||
:param file_hash: SHA256 hash to check
|
||||
:type file_hash: str
|
||||
:returns: Indicator details if matched, otherwise None
|
||||
|
||||
"""
|
||||
if not file_hash:
|
||||
return None
|
||||
|
||||
for ioc in self.get_iocs("files_sha256"):
|
||||
if file_hash.lower() == ioc["value"].lower():
|
||||
self.log.warning("Found a known suspicious file with hash \"%s\" matching indicators from \"%s\"",
|
||||
file_hash, ioc["name"])
|
||||
return ioc
|
||||
|
||||
def check_app_id(self, app_id):
|
||||
"""Check the provided app identifier (typically an Android package name)
|
||||
against the list of indicators.
|
||||
|
||||
:param app_id: App ID to check against the list of indicators
|
||||
:type app_id: str
|
||||
:returns: Indicator details if matched, otherwise None
|
||||
|
||||
"""
|
||||
if not app_id:
|
||||
return None
|
||||
|
||||
for ioc in self.get_iocs("app_ids"):
|
||||
if app_id.lower() == ioc["value"].lower():
|
||||
self.log.warning("Found a known suspicious app with ID \"%s\" matching indicators from \"%s\"",
|
||||
app_id, ioc["name"])
|
||||
return ioc
|
||||
|
||||
|
||||
def download_indicators_files(log):
|
||||
"""
|
||||
|
||||
@@ -6,7 +6,7 @@
|
||||
import requests
|
||||
from packaging import version
|
||||
|
||||
MVT_VERSION = "1.4.7"
|
||||
MVT_VERSION = "1.4.9"
|
||||
|
||||
|
||||
def check_for_updates():
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021 The MVT Project Authors.
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
@@ -78,6 +77,9 @@ class ConfigurationProfiles(IOSExtraction):
|
||||
|
||||
if "SignerCerts" in conf_plist:
|
||||
conf_plist["SignerCerts"] = [b64encode(x) for x in conf_plist["SignerCerts"]]
|
||||
if "OTAProfileStub" in conf_plist:
|
||||
if "SignerCerts" in conf_plist["OTAProfileStub"]:
|
||||
conf_plist["OTAProfileStub"]["SignerCerts"] = [b64encode(x) for x in conf_plist["OTAProfileStub"]["SignerCerts"]]
|
||||
if "PushTokenDataSentToServerKey" in conf_plist:
|
||||
conf_plist["PushTokenDataSentToServerKey"] = b64encode(conf_plist["PushTokenDataSentToServerKey"])
|
||||
if "LastPushTokenHash" in conf_plist:
|
||||
|
||||
@@ -62,6 +62,7 @@ class SafariBrowserState(IOSExtraction):
|
||||
self.detected.append(result)
|
||||
|
||||
def _process_browser_state_db(self, db_path):
|
||||
self._recover_sqlite_db_if_needed(db_path)
|
||||
conn = sqlite3.connect(db_path)
|
||||
|
||||
cur = conn.cursor()
|
||||
@@ -92,8 +93,12 @@ class SafariBrowserState(IOSExtraction):
|
||||
if row[4]:
|
||||
# Skip a 4 byte header before the plist content.
|
||||
session_plist = row[4][4:]
|
||||
session_data = plistlib.load(io.BytesIO(session_plist))
|
||||
session_data = keys_bytes_to_string(session_data)
|
||||
session_data = {}
|
||||
try:
|
||||
session_data = plistlib.load(io.BytesIO(session_plist))
|
||||
session_data = keys_bytes_to_string(session_data)
|
||||
except plistlib.InvalidFileException:
|
||||
pass
|
||||
|
||||
if "SessionHistoryEntries" in session_data.get("SessionHistory", {}):
|
||||
for session_entry in session_data["SessionHistory"].get("SessionHistoryEntries"):
|
||||
@@ -114,7 +119,6 @@ class SafariBrowserState(IOSExtraction):
|
||||
})
|
||||
|
||||
def run(self):
|
||||
|
||||
if self.is_backup:
|
||||
for backup_file in self._get_backup_files_from_manifest(relative_path=SAFARI_BROWSER_STATE_BACKUP_RELPATH):
|
||||
self.file_path = self._get_backup_file_from_id(backup_file["file_id"])
|
||||
|
||||
@@ -28,7 +28,7 @@ class WebkitResourceLoadStatistics(IOSExtraction):
|
||||
output_folder=output_folder, fast_mode=fast_mode,
|
||||
log=log, results=results)
|
||||
|
||||
self.results = {}
|
||||
self.results = {} if not results else results
|
||||
|
||||
def check_indicators(self):
|
||||
if not self.indicators:
|
||||
|
||||
@@ -35,7 +35,7 @@ class WebkitSessionResourceLog(IOSExtraction):
|
||||
output_folder=output_folder, fast_mode=fast_mode,
|
||||
log=log, results=results)
|
||||
|
||||
self.results = {}
|
||||
self.results = {} if not results else results
|
||||
|
||||
@staticmethod
|
||||
def _extract_domains(entries):
|
||||
|
||||
@@ -236,6 +236,7 @@ IPHONE_IOS_VERSIONS = [
|
||||
{"build": "19B81", "version": "15.1.1"},
|
||||
{"build": "19C56", "version": "15.2"},
|
||||
{"build": "19C63", "version": "15.2.1"},
|
||||
{"build": "19D50", "version": "15.3"},
|
||||
]
|
||||
|
||||
|
||||
|
||||
Binary file not shown.
36
tests/ios/test_safari_browserstate.py
Normal file
36
tests/ios/test_safari_browserstate.py
Normal file
@@ -0,0 +1,36 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021 The MVT Project Authors.
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import logging
|
||||
|
||||
from mvt.common.indicators import Indicators
|
||||
from mvt.common.module import run_module
|
||||
from mvt.ios.modules.mixed.safari_browserstate import SafariBrowserState
|
||||
|
||||
from ..utils import get_backup_folder
|
||||
|
||||
|
||||
class TestSafariBrowserStateModule:
|
||||
def test_parsing(self):
|
||||
m = SafariBrowserState(base_folder=get_backup_folder(), log=logging, results=[])
|
||||
m.is_backup = True
|
||||
run_module(m)
|
||||
assert m.file_path != None
|
||||
assert len(m.results) == 1
|
||||
assert len(m.timeline) == 1
|
||||
assert len(m.detected) == 0
|
||||
|
||||
def test_detection(self, indicator_file):
|
||||
m = SafariBrowserState(base_folder=get_backup_folder(), log=logging, results=[])
|
||||
m.is_backup = True
|
||||
ind = Indicators(log=logging)
|
||||
ind.parse_stix2(indicator_file)
|
||||
# Adds a file that exists in the manifest.
|
||||
ind.ioc_files[0]["domains"].append("en.wikipedia.org")
|
||||
m.indicators = ind
|
||||
run_module(m)
|
||||
assert len(m.detected) == 1
|
||||
assert len(m.results) == 1
|
||||
assert m.results[0]["tab_url"] == "https://en.wikipedia.org/wiki/NSO_Group"
|
||||
Reference in New Issue
Block a user