Ensuring all adb connect/disconnect are happening in modules only

This commit is contained in:
Nex 2022-08-13 23:12:43 +02:00
parent a966b694ea
commit 66a157868f
5 changed files with 39 additions and 15 deletions

View File

@ -164,8 +164,6 @@ class AndroidExtraction(MVTModule):
# TODO: Need to support checking files without root privileges as well.
# Connect to the device over adb.
self._adb_connect()
# Check if we have root, if not raise an Exception.
self._adb_root_or_die()
@ -236,7 +234,6 @@ class AndroidExtraction(MVTModule):
"""
# Connect to the device over adb.
self._adb_connect()
# Check if we have root, if not raise an Exception.
self._adb_root_or_die()
@ -263,8 +260,6 @@ class AndroidExtraction(MVTModule):
tmp.close()
# Delete the copy on /sdcard/.
self._adb_command(f"rm -f {new_remote_path}")
# Disconnect from the device.
self._adb_disconnect()
def _generate_backup(self, package_name: str) -> bytes:
self.log.warning("Please check phone and accept Android backup prompt. "

View File

@ -81,8 +81,12 @@ class ChromeHistory(AndroidExtraction):
len(self.results))
def run(self) -> None:
self._adb_connect()
try:
self._adb_process_file(os.path.join("/", CHROME_HISTORY_PATH),
self._parse_db)
except Exception as exc:
self.log.error(exc)
self._adb_disconnect()

View File

@ -83,9 +83,18 @@ class Packages(AndroidExtraction):
records = []
timestamps = [
{"event": "package_install", "timestamp": record["timestamp"]},
{"event": "package_first_install", "timestamp": record["first_install_time"]},
{"event": "package_last_update", "timestamp": record["last_update_time"]},
{
"event": "package_install",
"timestamp": record["timestamp"]
},
{
"event": "package_first_install",
"timestamp": record["first_install_time"]
},
{
"event": "package_last_update",
"timestamp": record["last_update_time"]
},
]
for timestamp in timestamps:
@ -93,7 +102,8 @@ class Packages(AndroidExtraction):
"timestamp": timestamp["timestamp"],
"module": self.__class__.__name__,
"event": timestamp["event"],
"data": f"{record['package_name']} (system: {record['system']}, third party: {record['third_party']})",
"data": f"{record['package_name']} (system: {record['system']},"
f" third party: {record['third_party']})",
})
return records
@ -101,7 +111,8 @@ class Packages(AndroidExtraction):
def check_indicators(self) -> None:
for result in self.results:
if result["package_name"] in ROOT_PACKAGES:
self.log.warning("Found an installed package related to rooting/jailbreaking: \"%s\"",
self.log.warning("Found an installed package related to "
"rooting/jailbreaking: \"%s\"",
result["package_name"])
self.detected.append(result)
continue
@ -132,7 +143,8 @@ class Packages(AndroidExtraction):
total_hashes = len(hashes)
detections = {}
for i in track(range(total_hashes), description=f"Looking up {total_hashes} files..."):
progress_desc = f"Looking up {total_hashes} files..."
for i in track(range(total_hashes), description=progress_desc):
try:
results = virustotal_lookup(hashes[i])
except VTNoKey:
@ -303,8 +315,10 @@ class Packages(AndroidExtraction):
dangerous_permissions_count += 1
if dangerous_permissions_count >= DANGEROUS_PERMISSIONS_THRESHOLD:
self.log.info("Third-party package \"%s\" requested %d potentially dangerous permissions",
result["package_name"], dangerous_permissions_count)
self.log.info("Third-party package \"%s\" requested %d "
"potentially dangerous permissions",
result["package_name"],
dangerous_permissions_count)
packages_to_lookup = []
for result in self.results:
@ -312,8 +326,9 @@ class Packages(AndroidExtraction):
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"])
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:
self.check_virustotal(packages_to_lookup)

View File

@ -136,6 +136,8 @@ class SMS(AndroidExtraction):
len(self.results))
def run(self) -> None:
self._adb_connect()
try:
if self._adb_check_file_exists(os.path.join("/", SMS_BUGLE_PATH)):
self.sms_db_type = 1
@ -145,6 +147,8 @@ class SMS(AndroidExtraction):
self.sms_db_type = 2
self._adb_process_file(os.path.join("/", SMS_MMSSMS_PATH),
self._parse_db)
self._adb_disconnect()
return
except InsufficientPrivileges:
pass
@ -152,3 +156,5 @@ class SMS(AndroidExtraction):
self.log.warn("No SMS database found. Trying extraction of SMS data "
"using Android backup feature.")
self._extract_sms_adb()
self._adb_disconnect()

View File

@ -89,8 +89,12 @@ class Whatsapp(AndroidExtraction):
self.results = messages
def run(self) -> None:
self._adb_connect()
try:
self._adb_process_file(os.path.join("/", WHATSAPP_PATH),
self._parse_db)
except Exception as exc:
self.log.error(exc)
self._adb_disconnect()