mirror of https://github.com/mvt-project/mvt.git
Standardizing reST docstrings
This commit is contained in:
parent
3236c1b390
commit
e5f2aa3c3d
|
@ -32,7 +32,10 @@ class PullProgress(tqdm):
|
|||
|
||||
class DownloadAPKs(AndroidExtraction):
|
||||
"""DownloadAPKs is the main class operating the download of APKs
|
||||
from the device."""
|
||||
from the device.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, output_folder=None, all_apks=False, log=None,
|
||||
packages=None):
|
||||
|
@ -51,7 +54,9 @@ class DownloadAPKs(AndroidExtraction):
|
|||
@classmethod
|
||||
def from_json(cls, json_path):
|
||||
"""Initialize this class from an existing apks.json file.
|
||||
|
||||
:param json_path: Path to the apks.json file to parse.
|
||||
|
||||
"""
|
||||
with open(json_path, "r") as handle:
|
||||
packages = json.load(handle)
|
||||
|
@ -59,9 +64,11 @@ class DownloadAPKs(AndroidExtraction):
|
|||
|
||||
def pull_package_file(self, package_name, remote_path):
|
||||
"""Pull files related to specific package from the device.
|
||||
|
||||
:param package_name: Name of the package to download
|
||||
:param remote_path: Path to the file to download
|
||||
:returns: Path to the local copy
|
||||
|
||||
"""
|
||||
log.info("Downloading %s ...", remote_path)
|
||||
|
||||
|
@ -101,6 +108,8 @@ class DownloadAPKs(AndroidExtraction):
|
|||
def get_packages(self):
|
||||
"""Use the Packages adb module to retrieve the list of packages.
|
||||
We reuse the same extraction logic to then download the APKs.
|
||||
|
||||
|
||||
"""
|
||||
self.log.info("Retrieving list of installed packages...")
|
||||
|
||||
|
@ -111,8 +120,7 @@ class DownloadAPKs(AndroidExtraction):
|
|||
self.packages = m.results
|
||||
|
||||
def pull_packages(self):
|
||||
"""Download all files of all selected packages from the device.
|
||||
"""
|
||||
"""Download all files of all selected packages from the device."""
|
||||
log.info("Starting extraction of installed APKs at folder %s", self.output_folder)
|
||||
|
||||
if not os.path.exists(self.output_folder):
|
||||
|
@ -185,15 +193,13 @@ class DownloadAPKs(AndroidExtraction):
|
|||
log.info("Download of selected packages completed")
|
||||
|
||||
def save_json(self):
|
||||
"""Save the results to the package.json file.
|
||||
"""
|
||||
"""Save the results to the package.json file."""
|
||||
json_path = os.path.join(self.output_folder, "apks.json")
|
||||
with open(json_path, "w") as handle:
|
||||
json.dump(self.packages, handle, indent=4)
|
||||
|
||||
def run(self):
|
||||
"""Run all steps of fetch-apk.
|
||||
"""
|
||||
"""Run all steps of fetch-apk."""
|
||||
self.get_packages()
|
||||
self._adb_connect()
|
||||
self.pull_packages()
|
||||
|
|
|
@ -39,8 +39,7 @@ class AndroidExtraction(MVTModule):
|
|||
|
||||
@staticmethod
|
||||
def _adb_check_keys():
|
||||
"""Make sure Android adb keys exist.
|
||||
"""
|
||||
"""Make sure Android adb keys exist."""
|
||||
if not os.path.isdir(os.path.dirname(ADB_KEY_PATH)):
|
||||
os.path.makedirs(os.path.dirname(ADB_KEY_PATH))
|
||||
|
||||
|
@ -51,8 +50,7 @@ class AndroidExtraction(MVTModule):
|
|||
write_public_keyfile(ADB_KEY_PATH, ADB_PUB_KEY_PATH)
|
||||
|
||||
def _adb_connect(self):
|
||||
"""Connect to the device over adb.
|
||||
"""
|
||||
"""Connect to the device over adb."""
|
||||
self._adb_check_keys()
|
||||
|
||||
with open(ADB_KEY_PATH, "rb") as handle:
|
||||
|
@ -94,47 +92,53 @@ class AndroidExtraction(MVTModule):
|
|||
break
|
||||
|
||||
def _adb_disconnect(self):
|
||||
"""Close adb connection to the device.
|
||||
"""
|
||||
"""Close adb connection to the device."""
|
||||
self.device.close()
|
||||
|
||||
def _adb_reconnect(self):
|
||||
"""Reconnect to device using adb.
|
||||
"""
|
||||
"""Reconnect to device using adb."""
|
||||
log.info("Reconnecting ...")
|
||||
self._adb_disconnect()
|
||||
self._adb_connect()
|
||||
|
||||
def _adb_command(self, command):
|
||||
"""Execute an adb shell command.
|
||||
|
||||
:param command: Shell command to execute
|
||||
:returns: Output of command
|
||||
|
||||
"""
|
||||
return self.device.shell(command)
|
||||
|
||||
def _adb_check_if_root(self):
|
||||
"""Check if we have a `su` binary on the Android device.
|
||||
|
||||
|
||||
:returns: Boolean indicating whether a `su` binary is present or not
|
||||
|
||||
"""
|
||||
return bool(self._adb_command("command -v su"))
|
||||
|
||||
def _adb_root_or_die(self):
|
||||
"""Check if we have a `su` binary, otherwise raise an Exception.
|
||||
"""
|
||||
"""Check if we have a `su` binary, otherwise raise an Exception."""
|
||||
if not self._adb_check_if_root():
|
||||
raise InsufficientPrivileges("This module is optionally available in case the device is already rooted. Do NOT root your own device!")
|
||||
|
||||
def _adb_command_as_root(self, command):
|
||||
"""Execute an adb shell command.
|
||||
|
||||
:param command: Shell command to execute as root
|
||||
:returns: Output of command
|
||||
|
||||
"""
|
||||
return self._adb_command(f"su -c {command}")
|
||||
|
||||
def _adb_check_file_exists(self, file):
|
||||
"""Verify that a file exists.
|
||||
|
||||
:param file: Path of the file
|
||||
:returns: Boolean indicating whether the file exists or not
|
||||
|
||||
"""
|
||||
|
||||
# TODO: Need to support checking files without root privileges as well.
|
||||
|
@ -148,9 +152,12 @@ class AndroidExtraction(MVTModule):
|
|||
|
||||
def _adb_download(self, remote_path, local_path, progress_callback=None, retry_root=True):
|
||||
"""Download a file form the device.
|
||||
|
||||
:param remote_path: Path to download from the device
|
||||
:param local_path: Path to where to locally store the copy of the file
|
||||
:param progress_callback: Callback for download progress bar
|
||||
:param progress_callback: Callback for download progress bar (Default value = None)
|
||||
:param retry_root: Default value = True)
|
||||
|
||||
"""
|
||||
try:
|
||||
self.device.pull(remote_path, local_path, progress_callback)
|
||||
|
@ -191,9 +198,11 @@ class AndroidExtraction(MVTModule):
|
|||
def _adb_process_file(self, remote_path, process_routine):
|
||||
"""Download a local copy of a file which is only accessible as root.
|
||||
This is a wrapper around process_routine.
|
||||
|
||||
:param remote_path: Path of the file on the device to process
|
||||
:param process_routine: Function to be called on the local copy of the
|
||||
downloaded file
|
||||
|
||||
"""
|
||||
# Connect to the device over adb.
|
||||
self._adb_connect()
|
||||
|
@ -227,6 +236,5 @@ class AndroidExtraction(MVTModule):
|
|||
self._adb_disconnect()
|
||||
|
||||
def run(self):
|
||||
"""Run the main procedure.
|
||||
"""
|
||||
"""Run the main procedure."""
|
||||
raise NotImplementedError
|
||||
|
|
|
@ -35,7 +35,9 @@ class ChromeHistory(AndroidExtraction):
|
|||
|
||||
def _parse_db(self, db_path):
|
||||
"""Parse a Chrome History database file.
|
||||
|
||||
:param db_path: Path to the History database to process.
|
||||
|
||||
"""
|
||||
conn = sqlite3.connect(db_path)
|
||||
cur = conn.cursor()
|
||||
|
|
|
@ -71,7 +71,9 @@ class SMS(AndroidExtraction):
|
|||
|
||||
def _parse_db(self, db_path):
|
||||
"""Parse an Android bugle_db SMS database file.
|
||||
|
||||
:param db_path: Path to the Android SMS database file to process
|
||||
|
||||
"""
|
||||
conn = sqlite3.connect(db_path)
|
||||
cur = conn.cursor()
|
||||
|
|
|
@ -48,7 +48,9 @@ class Whatsapp(AndroidExtraction):
|
|||
|
||||
def _parse_db(self, db_path):
|
||||
"""Parse an Android msgstore.db WhatsApp database file.
|
||||
|
||||
:param db_path: Path to the Android WhatsApp database file to process
|
||||
|
||||
"""
|
||||
conn = sqlite3.connect(db_path)
|
||||
cur = conn.cursor()
|
||||
|
|
|
@ -15,6 +15,8 @@ class IndicatorsFileBadFormat(Exception):
|
|||
class Indicators:
|
||||
"""This class is used to parse indicators from a STIX2 file and provide
|
||||
functions to compare extracted artifacts to the indicators.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, log=None):
|
||||
|
@ -32,6 +34,10 @@ class Indicators:
|
|||
|
||||
def parse_stix2(self, file_path):
|
||||
"""Extract indicators from a STIX2 file.
|
||||
|
||||
:param file_path: Path to the STIX2 file to parse
|
||||
:type file_path: str
|
||||
|
||||
"""
|
||||
self.log.info("Parsing STIX2 indicators file at path %s",
|
||||
file_path)
|
||||
|
@ -64,9 +70,20 @@ class Indicators:
|
|||
self._add_indicator(ioc=value,
|
||||
iocs_list=self.ioc_files)
|
||||
|
||||
def check_domain(self, url):
|
||||
def check_domain(self, url) -> bool:
|
||||
"""Check if a given URL matches any of the provided domain indicators.
|
||||
|
||||
:param url: URL to match against domain indicators
|
||||
:type url: str
|
||||
:returns: True if the URL matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
|
||||
# TODO: If the IOC domain contains a subdomain, it is not currently
|
||||
# being matched.
|
||||
if not url:
|
||||
return False
|
||||
|
||||
try:
|
||||
# First we use the provided URL.
|
||||
|
@ -124,18 +141,35 @@ class Indicators:
|
|||
|
||||
return True
|
||||
|
||||
def check_domains(self, urls):
|
||||
"""Check the provided list of (suspicious) domains against a list of URLs.
|
||||
:param urls: List of URLs to check
|
||||
return False
|
||||
|
||||
def check_domains(self, urls) -> bool:
|
||||
"""Check a list of URLs against the provided list of domain 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
|
||||
|
||||
"""
|
||||
if not urls:
|
||||
return False
|
||||
|
||||
for url in urls:
|
||||
if self.check_domain(url):
|
||||
return True
|
||||
|
||||
def check_process(self, process):
|
||||
return False
|
||||
|
||||
def check_process(self, process) -> bool:
|
||||
"""Check the provided process name against the list of process
|
||||
indicators.
|
||||
:param process: Process name to check
|
||||
|
||||
:param process: Process name to check against process indicators
|
||||
:type process: str
|
||||
:returns: True if process matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
if not process:
|
||||
return False
|
||||
|
@ -151,18 +185,35 @@ class Indicators:
|
|||
self.log.warning("Found a truncated known suspicious process name \"%s\"", process)
|
||||
return True
|
||||
|
||||
def check_processes(self, processes):
|
||||
return False
|
||||
|
||||
def check_processes(self, processes) -> bool:
|
||||
"""Check the provided list of processes against the list of
|
||||
process indicators.
|
||||
:param processes: List of processes to check
|
||||
|
||||
:param processes: List of processes to check against process indicators
|
||||
:type processes: list
|
||||
:returns: True if process matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
if not processes:
|
||||
return False
|
||||
|
||||
for process in processes:
|
||||
if self.check_process(process):
|
||||
return True
|
||||
|
||||
def check_email(self, email):
|
||||
return False
|
||||
|
||||
def check_email(self, email) -> bool:
|
||||
"""Check the provided email against the list of email indicators.
|
||||
:param email: Suspicious email to check
|
||||
|
||||
:param email: Email address to check against email indicators
|
||||
:type email: str
|
||||
:returns: True if email address matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
if not email:
|
||||
return False
|
||||
|
@ -171,9 +222,17 @@ class Indicators:
|
|||
self.log.warning("Found a known suspicious email address: \"%s\"", email)
|
||||
return True
|
||||
|
||||
def check_file(self, file_path):
|
||||
return False
|
||||
|
||||
def check_file(self, file_path) -> bool:
|
||||
"""Check the provided file path against the list of file indicators.
|
||||
:param file_path: Path or name of the file to check
|
||||
|
||||
: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
|
||||
|
||||
"""
|
||||
if not file_path:
|
||||
return False
|
||||
|
@ -182,3 +241,5 @@ class Indicators:
|
|||
if file_name in self.ioc_files:
|
||||
self.log.warning("Found a known suspicious file: \"%s\"", file_path)
|
||||
return True
|
||||
|
||||
return False
|
||||
|
|
|
@ -31,7 +31,7 @@ class MVTModule(object):
|
|||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
fast_mode=False, log=None, results=[]):
|
||||
"""Initialize module.
|
||||
:param file_path: Path to the module's database file, if there is any.
|
||||
:param file_path: Path to the module's database file, if there is any
|
||||
:param base_folder: Path to the base folder (backup or filesystem dump)
|
||||
:param output_folder: Folder where results will be stored
|
||||
:param fast_mode: Flag to enable or disable slow modules
|
||||
|
@ -70,12 +70,14 @@ class MVTModule(object):
|
|||
|
||||
def check_indicators(self):
|
||||
"""Check the results of this module against a provided list of
|
||||
indicators."""
|
||||
indicators.
|
||||
|
||||
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def save_to_json(self):
|
||||
"""Save the collected results to a json file.
|
||||
"""
|
||||
"""Save the collected results to a json file."""
|
||||
if not self.output_folder:
|
||||
return
|
||||
|
||||
|
@ -102,15 +104,18 @@ class MVTModule(object):
|
|||
|
||||
@staticmethod
|
||||
def _deduplicate_timeline(timeline):
|
||||
"""Serialize entry as JSON to deduplicate repeated entries"""
|
||||
"""Serialize entry as JSON to deduplicate repeated entries
|
||||
|
||||
:param timeline: List of entries from timeline to deduplicate
|
||||
|
||||
"""
|
||||
timeline_set = set()
|
||||
for record in timeline:
|
||||
timeline_set.add(json.dumps(record, sort_keys=True))
|
||||
return [json.loads(record) for record in timeline_set]
|
||||
|
||||
def to_timeline(self):
|
||||
"""Convert results into a timeline.
|
||||
"""
|
||||
"""Convert results into a timeline."""
|
||||
for result in self.results:
|
||||
record = self.serialize(result)
|
||||
if record:
|
||||
|
@ -132,8 +137,7 @@ class MVTModule(object):
|
|||
self.timeline_detected = self._deduplicate_timeline(self.timeline_detected)
|
||||
|
||||
def run(self):
|
||||
"""Run the main module procedure.
|
||||
"""
|
||||
"""Run the main module procedure."""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
|
@ -178,8 +182,10 @@ def run_module(module):
|
|||
|
||||
def save_timeline(timeline, timeline_path):
|
||||
"""Save the timeline in a csv file.
|
||||
:param timeline: List of records to order and store.
|
||||
:param timeline_path: Path to the csv file to store the timeline to.
|
||||
|
||||
:param timeline: List of records to order and store
|
||||
:param timeline_path: Path to the csv file to store the timeline to
|
||||
|
||||
"""
|
||||
with io.open(timeline_path, "a+", encoding="utf-8") as handle:
|
||||
csvoutput = csv.writer(handle, delimiter=",", quotechar="\"")
|
||||
|
|
|
@ -9,8 +9,7 @@ from click import Option, UsageError
|
|||
|
||||
|
||||
class MutuallyExclusiveOption(Option):
|
||||
"""This class extends click to support mutually exclusive options.
|
||||
"""
|
||||
"""This class extends click to support mutually exclusive options."""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
self.mutually_exclusive = set(kwargs.pop("mutually_exclusive", []))
|
||||
|
|
|
@ -263,8 +263,12 @@ class URL:
|
|||
|
||||
def get_domain(self):
|
||||
"""Get the domain from a URL.
|
||||
|
||||
:param url: URL to parse
|
||||
:returns: Just the domain name extracted from the URL
|
||||
:type url: str
|
||||
:returns: Domain name extracted from URL
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
# TODO: Properly handle exception.
|
||||
try:
|
||||
|
@ -273,9 +277,13 @@ class URL:
|
|||
return None
|
||||
|
||||
def get_top_level(self):
|
||||
"""Get only the top level domain from a URL.
|
||||
"""Get only the top-level domain from a URL.
|
||||
|
||||
:param url: URL to parse
|
||||
:returns: The top level domain extracted from the URL
|
||||
:type url: str
|
||||
:returns: Top-level domain name extracted from URL
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
# TODO: Properly handle exception.
|
||||
try:
|
||||
|
@ -283,13 +291,22 @@ class URL:
|
|||
except:
|
||||
return None
|
||||
|
||||
def check_if_shortened(self):
|
||||
def check_if_shortened(self) -> bool:
|
||||
"""Check if the URL is among list of shortener services.
|
||||
|
||||
|
||||
:returns: True if the URL is shortened, otherwise False
|
||||
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
if self.domain.lower() in SHORTENER_DOMAINS:
|
||||
self.is_shortened = True
|
||||
|
||||
return self.is_shortened
|
||||
|
||||
def unshorten(self):
|
||||
"""Unshorten the URL by requesting an HTTP HEAD response."""
|
||||
res = requests.head(self.url)
|
||||
if str(res.status_code).startswith("30"):
|
||||
return res.headers["Location"]
|
||||
|
|
|
@ -10,8 +10,13 @@ import re
|
|||
|
||||
def convert_mactime_to_unix(timestamp, from_2001=True):
|
||||
"""Converts Mac Standard Time to a Unix timestamp.
|
||||
:param timestamp: MacTime timestamp (either int or float)
|
||||
:returns: Unix epoch timestamp
|
||||
|
||||
:param timestamp: MacTime timestamp (either int or float).
|
||||
:type timestamp: int
|
||||
:param from_2001: bool: Whether to (Default value = True)
|
||||
:param from_2001: Default value = True)
|
||||
:returns: Unix epoch timestamp.
|
||||
|
||||
"""
|
||||
if not timestamp:
|
||||
return None
|
||||
|
@ -34,8 +39,11 @@ def convert_mactime_to_unix(timestamp, from_2001=True):
|
|||
|
||||
def convert_chrometime_to_unix(timestamp):
|
||||
"""Converts Chrome timestamp to a Unix timestamp.
|
||||
:param timestamp: Chrome timestamp as int
|
||||
:returns: Unix epoch timestamp
|
||||
|
||||
:param timestamp: Chrome timestamp as int.
|
||||
:type timestamp: int
|
||||
:returns: Unix epoch timestamp.
|
||||
|
||||
"""
|
||||
epoch_start = datetime.datetime(1601, 1 , 1)
|
||||
delta = datetime.timedelta(microseconds=timestamp)
|
||||
|
@ -44,8 +52,12 @@ def convert_chrometime_to_unix(timestamp):
|
|||
|
||||
def convert_timestamp_to_iso(timestamp):
|
||||
"""Converts Unix timestamp to ISO string.
|
||||
:param timestamp: Unix timestamp
|
||||
:returns: ISO timestamp string in YYYY-mm-dd HH:MM:SS.ms format
|
||||
|
||||
:param timestamp: Unix timestamp.
|
||||
:type timestamp: int
|
||||
:returns: ISO timestamp string in YYYY-mm-dd HH:MM:SS.ms format.
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
try:
|
||||
return timestamp.strftime("%Y-%m-%d %H:%M:%S.%f")
|
||||
|
@ -54,15 +66,20 @@ def convert_timestamp_to_iso(timestamp):
|
|||
|
||||
def check_for_links(text):
|
||||
"""Checks if a given text contains HTTP links.
|
||||
:param text: Any provided text
|
||||
:returns: Search results
|
||||
|
||||
:param text: Any provided text.
|
||||
:type text: str
|
||||
:returns: Search results.
|
||||
|
||||
"""
|
||||
return re.findall("(?P<url>https?://[^\s]+)", text, re.IGNORECASE)
|
||||
|
||||
def get_sha256_from_file_path(file_path):
|
||||
"""Calculate the SHA256 hash of a file from a file path.
|
||||
|
||||
:param file_path: Path to the file to hash
|
||||
:returns: The SHA256 hash string
|
||||
|
||||
"""
|
||||
sha256_hash = hashlib.sha256()
|
||||
with open(file_path, "rb") as handle:
|
||||
|
@ -75,8 +92,11 @@ def get_sha256_from_file_path(file_path):
|
|||
# https://stackoverflow.com/questions/57014259/json-dumps-on-dictionary-with-bytes-for-keys
|
||||
def keys_bytes_to_string(obj):
|
||||
"""Convert object keys from bytes to string.
|
||||
|
||||
:param obj: Object to convert from bytes to string.
|
||||
:returns: Converted object.
|
||||
:returns: Object converted to string.
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
new_obj = {}
|
||||
if not isinstance(obj, dict):
|
||||
|
|
|
@ -17,6 +17,8 @@ log = logging.getLogger(__name__)
|
|||
class DecryptBackup:
|
||||
"""This class provides functions to decrypt an encrypted iTunes backup
|
||||
using either a password or a key file.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, backup_path, dest_path=None):
|
||||
|
@ -35,7 +37,9 @@ class DecryptBackup:
|
|||
@staticmethod
|
||||
def is_encrypted(backup_path) -> bool:
|
||||
"""Query Manifest.db file to see if it's encrypted or not.
|
||||
|
||||
:param backup_path: Path to the backup to decrypt
|
||||
|
||||
"""
|
||||
conn = sqlite3.connect(os.path.join(backup_path, "Manifest.db"))
|
||||
cur = conn.cursor()
|
||||
|
@ -95,7 +99,9 @@ class DecryptBackup:
|
|||
|
||||
def decrypt_with_password(self, password):
|
||||
"""Decrypts an encrypted iOS backup.
|
||||
|
||||
:param password: Password to use to decrypt the original backup
|
||||
|
||||
"""
|
||||
log.info("Decrypting iOS backup at path %s with password", self.backup_path)
|
||||
|
||||
|
@ -131,7 +137,9 @@ class DecryptBackup:
|
|||
|
||||
def decrypt_with_key_file(self, key_file):
|
||||
"""Decrypts an encrypted iOS backup using a key file.
|
||||
|
||||
:param key_file: File to read the key bytes to decrypt the backup
|
||||
|
||||
"""
|
||||
log.info("Decrypting iOS backup at path %s with key file %s",
|
||||
self.backup_path, key_file)
|
||||
|
@ -158,8 +166,7 @@ class DecryptBackup:
|
|||
log.critical("Failed to decrypt backup. Did you provide the correct key file?")
|
||||
|
||||
def get_key(self):
|
||||
"""Retrieve and prints the encryption key.
|
||||
"""
|
||||
"""Retrieve and prints the encryption key."""
|
||||
if not self._backup:
|
||||
return
|
||||
|
||||
|
@ -169,7 +176,9 @@ class DecryptBackup:
|
|||
|
||||
def write_key(self, key_path):
|
||||
"""Save extracted key to file.
|
||||
|
||||
:param key_path: Path to the file where to write the derived decryption key.
|
||||
|
||||
"""
|
||||
if not self._decryption_key:
|
||||
return
|
||||
|
|
|
@ -11,8 +11,7 @@ from ..base import IOSExtraction
|
|||
CONF_PROFILES_DOMAIN = "SysSharedContainerDomain-systemgroup.com.apple.configurationprofiles"
|
||||
|
||||
class ConfigurationProfiles(IOSExtraction):
|
||||
"""This module extracts the full plist data from configuration profiles.
|
||||
"""
|
||||
"""This module extracts the full plist data from configuration profiles."""
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
fast_mode=False, log=None, results=[]):
|
||||
|
|
|
@ -27,12 +27,19 @@ class Manifest(IOSExtraction):
|
|||
def _get_key(self, dictionary, key):
|
||||
"""Unserialized plist objects can have keys which are str or byte types
|
||||
This is a helper to try fetch a key as both a byte or string type.
|
||||
|
||||
:param dictionary: param key:
|
||||
:param key:
|
||||
|
||||
"""
|
||||
return dictionary.get(key.encode("utf-8"), None) or dictionary.get(key, None)
|
||||
|
||||
@staticmethod
|
||||
def _convert_timestamp(timestamp_or_unix_time_int):
|
||||
"""Older iOS versions stored the manifest times as unix timestamps.
|
||||
|
||||
:param timestamp_or_unix_time_int:
|
||||
|
||||
"""
|
||||
if isinstance(timestamp_or_unix_time_int, datetime.datetime):
|
||||
return convert_timestamp_to_iso(timestamp_or_unix_time_int)
|
||||
|
|
|
@ -14,6 +14,8 @@ CONF_PROFILES_EVENTS_RELPATH = "Library/ConfigurationProfiles/MCProfileEvents.pl
|
|||
class ProfileEvents(IOSExtraction):
|
||||
"""This module extracts events related to the installation of configuration
|
||||
profiles.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
|
|
|
@ -28,7 +28,9 @@ class IOSExtraction(MVTModule):
|
|||
|
||||
def _recover_sqlite_db_if_needed(self, file_path):
|
||||
"""Tries to recover a malformed database by running a .clone command.
|
||||
|
||||
:param file_path: Path to the malformed database file.
|
||||
|
||||
"""
|
||||
# TODO: Find a better solution.
|
||||
conn = sqlite3.connect(file_path)
|
||||
|
@ -65,8 +67,10 @@ class IOSExtraction(MVTModule):
|
|||
|
||||
def _get_backup_files_from_manifest(self, relative_path=None, domain=None):
|
||||
"""Locate files from Manifest.db.
|
||||
:param relative_path: Relative path to use as filter from Manifest.db.
|
||||
:param domain: Domain to use as filter from Manifest.db.
|
||||
|
||||
:param relative_path: Relative path to use as filter from Manifest.db. (Default value = None)
|
||||
:param domain: Domain to use as filter from Manifest.db. (Default value = None)
|
||||
|
||||
"""
|
||||
manifest_db_path = os.path.join(self.base_folder, "Manifest.db")
|
||||
if not os.path.exists(manifest_db_path):
|
||||
|
@ -116,8 +120,11 @@ class IOSExtraction(MVTModule):
|
|||
modules that expect to work with a single SQLite database.
|
||||
If a module requires to process multiple databases or files,
|
||||
you should use the helper functions above.
|
||||
|
||||
:param backup_id: iTunes backup database file's ID (or hash).
|
||||
:param root_paths: Glob patterns for files to seek in filesystem dump.
|
||||
:param root_paths: Glob patterns for files to seek in filesystem dump. (Default value = [])
|
||||
:param backup_ids: Default value = None)
|
||||
|
||||
"""
|
||||
file_path = None
|
||||
# First we check if the was an explicit file path specified.
|
||||
|
|
|
@ -13,7 +13,10 @@ from ..base import IOSExtraction
|
|||
|
||||
class Filesystem(IOSExtraction):
|
||||
"""This module extracts creation and modification date of files from a
|
||||
full file-system dump."""
|
||||
full file-system dump.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
fast_mode=False, log=None, results=[]):
|
||||
|
|
|
@ -14,7 +14,10 @@ NETUSAGE_ROOT_PATHS = [
|
|||
|
||||
class Netusage(NetBase):
|
||||
"""This class extracts data from netusage.sqlite and attempts to identify
|
||||
any suspicious processes if running on a full filesystem dump."""
|
||||
any suspicious processes if running on a full filesystem dump.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
fast_mode=False, log=None, results=[]):
|
||||
|
|
|
@ -11,7 +11,10 @@ WEBKIT_INDEXEDDB_ROOT_PATHS = [
|
|||
|
||||
class WebkitIndexedDB(WebkitBase):
|
||||
"""This module looks extracts records from WebKit IndexedDB folders,
|
||||
and checks them against any provided list of suspicious domains."""
|
||||
and checks them against any provided list of suspicious domains.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
slug = "webkit_indexeddb"
|
||||
|
||||
|
|
|
@ -11,7 +11,10 @@ WEBKIT_LOCALSTORAGE_ROOT_PATHS = [
|
|||
|
||||
class WebkitLocalStorage(WebkitBase):
|
||||
"""This module looks extracts records from WebKit LocalStorage folders,
|
||||
and checks them against any provided list of suspicious domains."""
|
||||
and checks them against any provided list of suspicious domains.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
fast_mode=False, log=None, results=[]):
|
||||
|
|
|
@ -11,7 +11,10 @@ WEBKIT_SAFARIVIEWSERVICE_ROOT_PATHS = [
|
|||
|
||||
class WebkitSafariViewService(WebkitBase):
|
||||
"""This module looks extracts records from WebKit LocalStorage folders,
|
||||
and checks them against any provided list of suspicious domains."""
|
||||
and checks them against any provided list of suspicious domains.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
fast_mode=False, log=None, results=[]):
|
||||
|
|
|
@ -19,7 +19,10 @@ FIREFOX_HISTORY_ROOT_PATHS = [
|
|||
|
||||
class FirefoxHistory(IOSExtraction):
|
||||
"""This module extracts all Firefox visits and tries to detect potential
|
||||
network injection attacks."""
|
||||
network injection attacks.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
fast_mode=False, log=None, results=[]):
|
||||
|
|
|
@ -14,7 +14,10 @@ DATAUSAGE_ROOT_PATHS = [
|
|||
|
||||
class Datausage(NetBase):
|
||||
"""This class extracts data from DataUsage.sqlite and attempts to identify
|
||||
any suspicious processes if running on a full filesystem dump."""
|
||||
any suspicious processes if running on a full filesystem dump.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
fast_mode=False, log=None, results=[]):
|
||||
|
|
|
@ -19,7 +19,10 @@ SAFARI_HISTORY_ROOT_PATHS = [
|
|||
|
||||
class SafariHistory(IOSExtraction):
|
||||
"""This module extracts all Safari visits and tries to detect potential
|
||||
network injection attacks."""
|
||||
network injection attacks.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
fast_mode=False, log=None, results=[]):
|
||||
|
|
|
@ -18,8 +18,7 @@ WEBKIT_RESOURCELOADSTATICS_ROOT_PATHS = [
|
|||
]
|
||||
|
||||
class WebkitResourceLoadStatistics(IOSExtraction):
|
||||
"""This module extracts records from WebKit ResourceLoadStatistics observations.db.
|
||||
"""
|
||||
"""This module extracts records from WebKit ResourceLoadStatistics observations.db."""
|
||||
# TODO: Add serialize().
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
|
|
|
@ -23,7 +23,10 @@ WEBKIT_SESSION_RESOURCE_LOG_ROOT_PATHS = [
|
|||
class WebkitSessionResourceLog(IOSExtraction):
|
||||
"""This module extracts records from WebKit browsing session
|
||||
resource logs, and checks them against any provided list of
|
||||
suspicious domains."""
|
||||
suspicious domains.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, file_path=None, base_folder=None, output_folder=None,
|
||||
fast_mode=False, log=None, results=[]):
|
||||
|
|
|
@ -151,8 +151,7 @@ class NetBase(IOSExtraction):
|
|||
self.log.warning(msg)
|
||||
|
||||
def check_manipulated(self):
|
||||
"""Check for missing or manipulate DB entries
|
||||
"""
|
||||
"""Check for missing or manipulate DB entries"""
|
||||
# Don't show duplicates for each missing process.
|
||||
missing_process_cache = set()
|
||||
for result in sorted(self.results, key=operator.itemgetter("live_isodate")):
|
||||
|
|
Loading…
Reference in New Issue