[4.2.8] Multiple APK Analysis improvements, general Code QA & bug fixes (#2470)

* Dockerfile QA
* Add sdk-build-tools to Docker image
* Replace biplist with plistlib std lib
* Fixed a bug in iOS pbxproj parsing
* Added support for APK parsing with aapt2/aapt
* Use aapt/aapt2 as a fallback for APK parsing, files listing and string extraction
* Added "started at" to Scan task queue model #2463
* Tasks List API to return string status #2464
* Replaced all minidom calls with defusedxml.minidom
* Code QA on android manifest data extraction and parsing
* Improved android file analysis
* Improved android manifest data extraction
* Improved android icon file extraction
* Improved android app name extraction
* Improved android appstore package details extraction
* Android string extraction to fallback on aapt2 strings
* APK analysis arguments refactor
* Handle packed APKs, refactor unzip to handle malformed APK files
* Handle reserved filename conflict during ZIP extraction
* Explicit Zipslip handling during ZIP extraction
* Graceful files extraction on unzip failure
* Removed bail out and continue analysis
* Moved androguard parsing to the start of static analysis
* AndroidManifest.xml fallback from apktool to androguard during extraction and parsing
* Updated Tasks UI to show started at
This commit is contained in:
Ajin Abraham
2024-11-28 17:26:50 -08:00
committed by GitHub
parent a015df5871
commit 5ce7e237af
40 changed files with 1026 additions and 595 deletions
+1 -4
View File
@@ -20,10 +20,6 @@ ENV DEBIAN_FRONTEND=noninteractive \
USER_ID=9901 \
MOBSF_PLATFORM=docker \
MOBSF_ADB_BINARY=/usr/bin/adb \
JDK_FILE=openjdk-22.0.2_linux-x64_bin.tar.gz \
JDK_FILE_ARM=openjdk-22.0.2_linux-aarch64_bin.tar.gz \
WKH_FILE=wkhtmltox_0.12.6.1-3.bookworm_amd64.deb \
WKH_FILE_ARM=wkhtmltox_0.12.6.1-3.bookworm_arm64.deb \
JAVA_HOME=/jdk-22.0.2 \
PATH=/jdk-22.0.2/bin:/root/.local/bin:$PATH \
DJANGO_SUPERUSER_USERNAME=mobsf \
@@ -32,6 +28,7 @@ ENV DEBIAN_FRONTEND=noninteractive \
# See https://docs.docker.com/develop/develop-images/dockerfile_best-practices/#run
RUN apt update -y && \
apt install -y --no-install-recommends \
android-sdk-build-tools \
android-tools-adb \
build-essential \
curl \
+4 -4
View File
@@ -19,11 +19,11 @@ from mobsf.MobSF.utils import (
read_sqlite,
)
from biplist import (
writePlistToString,
from plistlib import (
FMT_XML,
dumps,
)
logger = logging.getLogger(__name__)
@@ -57,7 +57,7 @@ def view_file(request, api=False):
return print_n_send_error_response(request, err, api)
dat = sfile.read_text('ISO-8859-1')
if fil.endswith('.plist') and dat.startswith('bplist0'):
dat = writePlistToString(dat).decode('utf-8', 'ignore')
dat = dumps(dat, fmt=FMT_XML).decode('utf-8', 'ignore')
if fil.endswith(('.xml', '.plist')) and typ in ['xml', 'plist']:
rtyp = 'xml'
elif typ == 'db':
+1 -1
View File
@@ -18,7 +18,7 @@ from mobsf.install.windows.setup import windows_config_local
logger = logging.getLogger(__name__)
VERSION = '4.2.7'
VERSION = '4.2.8'
BANNER = r"""
__ __ _ ____ _____ _ _ ____
| \/ | ___ | |__/ ___|| ___|_ _| || | |___ \
+28 -1
View File
@@ -6,10 +6,11 @@ import re
import sys
from shutil import which
from pathlib import Path
from platform import system
from concurrent.futures import ThreadPoolExecutor
from mobsf.MobSF.utils import (
find_aapt,
find_java_binary,
gen_sha256_hash,
get_adb,
@@ -72,9 +73,21 @@ def get_executable_hashes():
downloaded_tools,
manage_py,
]
aapt = 'aapt'
aapt2 = 'aapt2'
if system() == 'Windows':
aapt = 'aapt.exe'
aapt2 = 'aapt2.exe'
aapts = [find_aapt(aapt), find_aapt(aapt2)]
exec_loc.extend(Path(a) for a in aapts if a)
# External binaries used directly by MobSF
system_bins = [
'aapt',
'aapt.exe',
'aapt2',
'aapt2.exe',
'adb',
'adb.exe',
'which',
'wkhtmltopdf',
'httptools',
@@ -110,6 +123,8 @@ def get_executable_hashes():
settings.CLASSDUMP_BINARY,
settings.CLASSDUMP_SWIFT_BINARY,
getattr(settings, 'BUNDLE_TOOL', ''),
getattr(settings, 'AAPT2_BINARY', ''),
getattr(settings, 'AAPT_BINARY', ''),
]
for ubin in user_defined_bins:
if ubin:
@@ -222,3 +237,15 @@ def sanitize_filename(filename):
# Remove leading and trailing underscores
safe_filename = safe_filename.strip('_')
return safe_filename
def sanitize_for_logging(filename: str, max_length: int = 255) -> str:
"""Sanitize a filename to prevent log injection."""
# Remove newline, carriage return, and other risky characters
filename = filename.replace('\n', '_').replace('\r', '_').replace('\t', '_')
# Allow only safe characters (alphanumeric, underscore, dash, and period)
filename = re.sub(r'[^a-zA-Z0-9._-]', '_', filename)
# Truncate filename to the maximum allowed length
return filename[:max_length]
+2
View File
@@ -462,6 +462,8 @@ else:
VD2SVG_BINARY = os.getenv('MOBSF_VD2SVG_BINARY', '')
APKTOOL_BINARY = os.getenv('MOBSF_APKTOOL_BINARY', '')
ADB_BINARY = os.getenv('MOBSF_ADB_BINARY', '')
AAPT2_BINARY = os.getenv('MOBSF_AAPT2_BINARY', '')
AAPT_BINARY = os.getenv('MOBSF_AAPT_BINARY', '')
# iOS 3P Tools
JTOOL_BINARY = os.getenv('MOBSF_JTOOL_BINARY', '')
+29
View File
@@ -184,6 +184,32 @@ def find_java_binary():
return 'java'
def find_aapt(tool_name):
"""Find the specified tool (aapt or aapt2)."""
# Check system PATH for the tool
tool_path = shutil.which(tool_name)
if tool_path:
return tool_path
# Check common Android SDK locations
home_dir = Path.home() # Get the user's home directory
sdk_paths = [
home_dir / 'Library' / 'Android' / 'sdk', # macOS
home_dir / 'Android' / 'Sdk', # Linux
home_dir / 'AppData' / 'Local' / 'Android' / 'Sdk', # Windows
]
for sdk_path in sdk_paths:
build_tools_path = sdk_path / 'build-tools'
if build_tools_path.exists():
for version in sorted(build_tools_path.iterdir(), reverse=True):
tool_path = version / tool_name
if tool_path.exists():
return str(tool_path)
return None
def print_n_send_error_response(request,
msg,
api=False,
@@ -667,6 +693,8 @@ def common_check(instance_id):
def is_path_traversal(user_input):
"""Check for path traversal."""
if not user_input:
return False
if (('../' in user_input)
or ('%2e%2e' in user_input)
or ('..' in user_input)
@@ -836,6 +864,7 @@ def get_android_dm_exception_msg():
def get_android_src_dir(app_dir, typ):
"""Get Android source code location."""
src = None
if typ == 'apk':
src = app_dir / 'java_source'
elif typ == 'studio':
+1
View File
@@ -177,6 +177,7 @@ class EnqueuedTask(models.Model):
file_name = models.CharField(max_length=255)
created_at = models.DateTimeField(default=timezone.now)
status = models.CharField(max_length=255, default='Enqueued')
started_at = models.DateTimeField(null=True)
completed_at = models.DateTimeField(null=True)
app_name = models.CharField(max_length=255, default='')
@@ -1,7 +1,8 @@
# -*- coding: utf_8 -*-
# flake8: noqa
import os
from xml.dom import minidom
from defusedxml.minidom import parseString
_public_res = None
# copy the newest sdk/platforms/android-?/data/res/values/public.xml here
@@ -11,7 +12,7 @@ if _public_res is None:
xmlfile = os.path.join(root, "public.xml")
if os.path.isfile(xmlfile):
with open(xmlfile, "r") as fp:
_xml = minidom.parseString(fp.read())
_xml = parseString(fp.read())
for element in _xml.getElementsByTagName("public"):
_type = element.getAttribute('type')
_name = element.getAttribute('name')
+151
View File
@@ -0,0 +1,151 @@
# -*- coding: utf_8 -*-
"""Use aapt2 to extract APK features."""
import re
import logging
import subprocess
from platform import system
from pathlib import Path
from django.conf import settings
from mobsf.MobSF.utils import (
find_aapt,
)
logger = logging.getLogger(__name__)
class AndroidAAPT:
def __init__(self, apk_path):
self.aapt2_path = None
self.aapt_path = None
self.apk_path = apk_path
self.data = {
'permissions': [],
'uses_features': {},
'package': None,
'application_label': None,
'application_icon': None,
'launchable_activity': None,
'min_sdk_version': None,
'target_sdk_version': None,
}
# Check for custom AAPT2 path in settings
if (getattr(settings, 'AAPT2_BINARY', '')
and len(settings.AAPT2_BINARY) > 0
and Path(settings.AAPT2_BINARY).exists()):
self.aapt2_path = settings.AAPT2_BINARY
else:
aapt2 = 'aapt2.exe' if system() == 'Windows' else 'aapt2'
self.aapt2_path = find_aapt(aapt2)
# Check for custom AAPT path in settings
if (getattr(settings, 'AAPT_BINARY', '')
and len(settings.AAPT_BINARY) > 0
and Path(settings.AAPT_BINARY).exists()):
self.aapt_path = settings.AAPT_BINARY
else:
aapt = 'aapt.exe' if system() == 'Windows' else 'aapt'
self.aapt_path = find_aapt(aapt)
# Ensure both aapt and aapt2 are found
if not (self.aapt2_path and self.aapt_path):
raise FileNotFoundError('aapt and aapt2 found')
def _execute_command(self, args):
try:
out = subprocess.check_output(
args,
stderr=subprocess.STDOUT)
return out.decode('utf-8', errors='ignore')
except subprocess.CalledProcessError as e:
logger.warning(e.output)
return None
def _get_strings(self, output):
# Regex to match strings while ignoring paths (strings without slashes)
pattern = r'String #[\d]+ : ([^\/\n]+)'
matches = re.findall(pattern, output)
# Strip whitespace and return the extracted strings
return [match.strip() for match in matches]
def _parse_badging(self, output):
# Match the package information
package_match = re.search(r'package: name=\'([\w\.]+)\'', output)
if package_match:
self.data['package'] = package_match.group(1)
# Match permissions
permissions = re.findall(r'uses-permission: name=\'([\w\.]+)\'', output)
if permissions:
self.data['permissions'] = permissions
# Match minSdkVersion
min_sdk_match = re.search(r'minSdkVersion:\'(\d+)\'', output)
if min_sdk_match:
self.data['min_sdk_version'] = min_sdk_match.group(1)
# Match targetSdkVersion
target_sdk_match = re.search(r'targetSdkVersion:\'(\d+)\'', output)
if target_sdk_match:
self.data['target_sdk_version'] = target_sdk_match.group(1)
# Match application label
label_match = re.search(r'application-label(?:-[\w\-]+)?:\'([^\']+)\'', output)
if label_match:
self.data['application_label'] = label_match.group(1)
# Match application icon
icon_match = re.search(r'application:.*icon=\'([^\']+)\'', output)
if icon_match:
self.data['application_icon'] = icon_match.group(1)
# Match launchable activity
activity_match = re.search(r'launchable-activity: name=\'([\w\.]+)\'', output)
if activity_match:
self.data['launchable_activity'] = activity_match.group(1)
# Match used features
features = {}
feature_matches = re.findall(
(r'(uses-feature(?:-not-required)?|uses-implied-feature): '
r'name=\'([\w\.]+)\'(?: reason=\'([^\']+)\')?'),
output,
)
for feature_type, feature_name, reason in feature_matches:
features[feature_name] = {
'type': feature_type,
# e.g., 'uses-feature',
# 'uses-feature-not-required',
# 'uses-implied-feature'
'reason': reason if reason else 'No reason provided',
}
self.data['uses_features'] = features
return self.data
def get_apk_files(self):
"""List all files in the APK."""
output = self._execute_command(
[self.aapt_path, 'list', self.apk_path])
if output:
return output.splitlines()
return []
def get_apk_strings(self):
"""Extract strings from the APK."""
output = self._execute_command(
[self.aapt2_path, 'dump', 'strings', self.apk_path])
if output:
return self._get_strings(output)
return []
def get_apk_features(self):
"""Extract features from the APK."""
output = self._execute_command(
[self.aapt2_path, 'dump', 'badging', self.apk_path])
if output:
return self._parse_badging(output)
return self.data
+67 -102
View File
@@ -1,4 +1,4 @@
# -*- coding: utf_8 -*-
"""Android APK and Source Analysis."""
import logging
import shutil
@@ -29,8 +29,9 @@ from mobsf.StaticAnalyzer.views.common.binary.lib_analysis import (
library_analysis,
)
from mobsf.StaticAnalyzer.views.android.app import (
get_app_name,
parse_apk,
aapt_parse,
androguard_parse,
get_apk_name,
)
from mobsf.StaticAnalyzer.views.android.cert_analysis import (
cert_info,
@@ -53,10 +54,12 @@ from mobsf.StaticAnalyzer.views.android.manifest_analysis import (
manifest_analysis,
)
from mobsf.StaticAnalyzer.views.android.manifest_utils import (
get_manifest,
manifest_data,
extract_manifest_data,
get_parsed_manifest,
)
from mobsf.StaticAnalyzer.views.android.playstore import (
get_app_details,
)
from mobsf.StaticAnalyzer.views.android.playstore import get_app_details
from mobsf.StaticAnalyzer.views.android.strings import (
get_strings_metadata,
)
@@ -73,6 +76,7 @@ from mobsf.StaticAnalyzer.views.common.appsec import (
)
from mobsf.StaticAnalyzer.views.common.async_task import (
async_analysis,
enqueued_task_init,
update_enqueued_task,
)
from mobsf.MobSF.views.authorization import (
@@ -96,39 +100,27 @@ def get_size_and_hashes(app_dic):
app_dic['sha1'], app_dic['sha256'] = hash_gen(app_dic['md5'], app_dic['app_path'])
def get_manifest_data(checksum, app_dic, andro_apk=None):
def get_manifest_data(app_dic):
"""Get Manifest Data."""
# Manifest XML
mani_file, ns, mani_xml = get_manifest(
checksum,
app_dic['app_path'],
app_dic['app_dir'],
app_dic['tools_dir'],
app_dic['zipped'],
andro_apk,
)
app_dic['manifest_file'] = mani_file
app_dic['parsed_xml'] = mani_xml
# Manifest XML parsed
get_parsed_manifest(app_dic)
# Populates manifest_file, manifest_namespace, manifest_parsed_xml
# Manifest data extraction
man_data = manifest_data(
checksum,
app_dic['parsed_xml'],
ns)
man_data_dic = extract_manifest_data(app_dic)
# Manifest Analysis
man_analysis = manifest_analysis(
checksum,
app_dic['parsed_xml'],
ns,
man_data,
app_dic['zipped'],
app_dic['app_dir'])
return man_data, man_analysis
man_analysis = manifest_analysis(app_dic, man_data_dic)
return man_data_dic, man_analysis
def print_scan_subject(checksum, app_dic, man_data):
def print_scan_subject(app_dic, man_data):
"""Log scan subject."""
app_name = app_dic['real_name']
pkg_name = man_data['packagename']
checksum = app_dic['md5']
app_name = app_dic.get('real_name')
pkg_name = man_data.get('packagename')
pkg_name2 = app_dic.get('apk_features', {}).get('package')
if not pkg_name:
pkg_name = pkg_name2
subject = 'Android App'
if app_name and pkg_name:
subject = f'{app_name} ({pkg_name})'
@@ -141,7 +133,13 @@ def print_scan_subject(checksum, app_dic, man_data):
append_scan_status(checksum, msg)
if subject == 'Failed':
subject = f'({subject})'
return subject
app_dic['subject'] = subject
def clean_up(app_dic):
"""Clean up for pickling."""
app_dic['androguard_apk'] = None
app_dic['androguard_apk_resources'] = None
def apk_analysis_task(checksum, app_dic, rescan, queue=False):
@@ -150,64 +148,42 @@ def apk_analysis_task(checksum, app_dic, rescan, queue=False):
try:
if queue:
settings.ASYNC_ANALYSIS = True
enqueued_task_init(checksum)
append_scan_status(checksum, 'init')
get_size_and_hashes(app_dic)
msg = 'Extracting APK'
logger.info(msg)
append_scan_status(checksum, msg)
app_dic['zipped'] = 'apk'
# Extract APK and get files
app_dic['files'] = unzip(
checksum,
app_dic['app_path'],
app_dic['app_dir'])
logger.info('APK Extracted')
if not app_dic['files']:
# Can't Analyze APK, bail out.
msg = 'APK file is invalid or corrupt'
logger.error(msg)
append_scan_status(checksum, msg)
if queue:
return update_enqueued_task(
checksum, 'Failed', msg)
return context, msg
app_dic['zipped'] = 'apk'
app_dic['certz'] = get_hardcoded_cert_keystore(
checksum,
app_dic['files'])
# Parse APK with Androguard4
andro_apk = parse_apk(
checksum,
app_dic['app_path'])
# Extract APK data with Androguard
androguard_parse(app_dic)
# Populates androguard_apk, androguard_manifest_xml, androguard_string_resources
# Extract APK data with AAPT/AAPT2
aapt_parse(app_dic) # Populates apk_features, files, apk_strings
get_hardcoded_cert_keystore(app_dic) # Populates file_analysis
# Manifest Data
man_data, man_analysis = get_manifest_data(
checksum,
app_dic,
andro_apk)
man_data_dic, man_analysis = get_manifest_data(app_dic)
# Get App name
app_dic['real_name'] = get_app_name(
andro_apk,
app_dic['app_dir'],
True)
# Print scan subject
subject = print_scan_subject(checksum, app_dic, man_data)
app_dic['playstore'] = get_app_details(
checksum,
man_data['packagename'])
get_apk_name(app_dic) # Populates real_name
print_scan_subject(app_dic, man_data_dic) # Populate subject
get_app_details(app_dic, man_data_dic) # Populates playstore
# Malware Permission check
mal_perms = permissions.check_malware_permission(
checksum,
man_data['perm'])
man_data_dic['perm'])
man_analysis['malware_permissions'] = mal_perms
# Get icon
# apktool should run before this
get_icon_apk(andro_apk, app_dic)
get_icon_apk(app_dic) # Populates icon_path
elf_dict = library_analysis(
checksum,
app_dic['app_dir'],
'elf')
cert_dic = cert_info(
andro_apk,
app_dic,
man_data)
cert_dic = cert_info(app_dic, man_data_dic)
apkid_results = apkid.apkid_analysis(
checksum,
app_dic['app_path'])
@@ -229,14 +205,11 @@ def apk_analysis_task(checksum, app_dic, rescan, queue=False):
app_dic['app_dir'],
app_dic['zipped'],
app_dic['manifest_file'],
man_data['perm'])
man_data_dic['perm'])
# Get the strings and metadata
get_strings_metadata(
checksum,
andro_apk,
app_dic['app_dir'],
app_dic,
elf_dict['elf_strings'],
app_dic['zipped'],
['.java'],
code_an_dic)
# Firebase DB Check
@@ -249,7 +222,7 @@ def apk_analysis_task(checksum, app_dic, rescan, queue=False):
code_an_dic['urls_list'])
context = save_get_ctx(
app_dic,
man_data,
man_data_dic,
man_analysis,
code_an_dic,
cert_dic,
@@ -260,13 +233,16 @@ def apk_analysis_task(checksum, app_dic, rescan, queue=False):
)
if queue:
return update_enqueued_task(
checksum, subject, 'Success')
checksum, app_dic['subject'], 'Success')
return context, None
except Exception as exp:
if queue:
return update_enqueued_task(
checksum, 'Failed', repr(exp))
return context, repr(exp)
finally:
# Clean up
clean_up(app_dic)
def generate_dynamic_context(request, app_dic, checksum, context, api):
@@ -312,6 +288,7 @@ def src_analysis_task(checksum, app_dic, rescan, pro_type, queue=False):
try:
if queue:
settings.ASYNC_ANALYSIS = True
enqueued_task_init(checksum)
cert_dic = {
'certificate_info': '',
'certificate_status': '',
@@ -321,45 +298,33 @@ def src_analysis_task(checksum, app_dic, rescan, pro_type, queue=False):
app_dic['secrets'] = []
# Above fields are only available for APK and not ZIP
app_dic['zipped'] = pro_type
app_dic['certz'] = get_hardcoded_cert_keystore(
checksum,
app_dic['files'])
get_hardcoded_cert_keystore(app_dic)
# Manifest Data
man_data, man_analysis = get_manifest_data(
checksum,
app_dic)
man_data_dic, man_analysis = get_manifest_data(app_dic)
# Get app name
app_dic['real_name'] = get_app_name(
None,
app_dic['app_dir'],
False)
get_apk_name(app_dic)
# Print scan subject
subject = print_scan_subject(checksum, app_dic, man_data)
app_dic['playstore'] = get_app_details(
checksum,
man_data['packagename'])
print_scan_subject(app_dic, man_data_dic)
get_app_details(app_dic, man_data_dic)
# Malware Permission check
mal_perms = permissions.check_malware_permission(
checksum,
man_data['perm'])
man_data_dic['perm'])
man_analysis['malware_permissions'] = mal_perms
# Get icon
get_icon_from_src(
app_dic,
man_data['icons'])
man_data_dic['icons'])
code_an_dic = code_analysis(
checksum,
app_dic['app_dir'],
app_dic['zipped'],
app_dic['manifest_file'],
man_data['perm'])
man_data_dic['perm'])
# Get the strings and metadata
get_strings_metadata(
checksum,
app_dic,
None,
app_dic['app_dir'],
None,
app_dic['zipped'],
['.java', '.kt'],
code_an_dic)
# Firebase DB Check
@@ -378,7 +343,7 @@ def src_analysis_task(checksum, app_dic, rescan, pro_type, queue=False):
code_an_dic['domains'], [])
context = save_get_ctx(
app_dic,
man_data,
man_data_dic,
man_analysis,
code_an_dic,
cert_dic,
@@ -389,7 +354,7 @@ def src_analysis_task(checksum, app_dic, rescan, pro_type, queue=False):
)
if queue:
return update_enqueued_task(
checksum, subject, 'Success')
checksum, app_dic['subject'], 'Success')
except Exception as exp:
if queue:
return update_enqueued_task(
+106 -34
View File
@@ -8,6 +8,9 @@ from pathlib import Path
from mobsf.StaticAnalyzer.tools.androguard4 import (
apk,
)
from mobsf.StaticAnalyzer.views.android import (
aapt,
)
from mobsf.MobSF.utils import (
append_scan_status,
)
@@ -15,53 +18,122 @@ from mobsf.MobSF.utils import (
logger = logging.getLogger(__name__)
def parse_apk(checksum, app_path):
"""Androguard4 APK."""
def aapt_parse(app_dict):
"""Extract features from APK using aapt/aapt2."""
checksum = app_dict['md5']
app_dict['apk_features'] = {}
app_dict['apk_strings'] = []
try:
msg = 'Extracting APK features using aapt/aapt2'
logger.info(msg)
append_scan_status(checksum, msg)
aapt_obj = aapt.AndroidAAPT(app_dict['app_path'])
app_dict['apk_features'] = aapt_obj.get_apk_features()
if not app_dict.get('files'):
app_dict['files'] = aapt_obj.get_apk_files()
app_dict['apk_strings'] = aapt_obj.get_apk_strings()
except FileNotFoundError:
msg = 'aapt and aapt2 not found, skipping APK feature extraction'
logger.warning(msg)
append_scan_status(checksum, msg)
except Exception as exp:
msg = 'Failed to extract APK features using aapt/aapt2'
logger.warning(msg)
append_scan_status(checksum, msg, repr(exp))
def androguard_parse(app_dict):
"""Extract features from APK using androguard."""
checksum = app_dict['md5']
app_dict['androguard_apk'] = None
app_dict['androguard_manifest_xml'] = None
app_dict['androguard_apk_resources'] = None
app_dict['androguard_apk_name'] = None
app_dict['androguard_apk_icon'] = None
try:
msg = 'Parsing APK with androguard'
logger.info(msg)
append_scan_status(checksum, msg)
return apk.APK(app_path)
a = apk.APK(app_dict['app_path'])
if not a:
msg = 'Failed to parse APK with androguard'
logger.warning(msg)
append_scan_status(checksum, msg)
return
app_dict['androguard_apk'] = a
try:
app_dict['androguard_apk_name'] = a.get_app_name()
except Exception as exp:
msg = 'Failed to get app name with androguard'
logger.warning(msg)
append_scan_status(checksum, msg, repr(exp))
try:
app_dict['androguard_apk_icon'] = a.get_app_icon(max_dpi=0xFFFE - 1)
except Exception as exp:
msg = 'Failed to get app icon with androguard'
logger.warning(msg)
append_scan_status(checksum, msg, repr(exp))
try:
xml = a.get_android_manifest_axml().get_xml()
app_dict['androguard_manifest_xml'] = xml
except Exception as exp:
msg = 'Failed to parse AndroidManifest.xml with androguard'
logger.warning(msg)
append_scan_status(checksum, msg, repr(exp))
try:
app_dict['androguard_apk_resources'] = a.get_android_resources()
except Exception as exp:
msg = 'Failed to parse resources with androguard'
logger.warning(msg)
append_scan_status(checksum, msg, repr(exp))
except Exception as exp:
msg = 'Failed to parse APK with androguard'
logger.error(msg)
append_scan_status(checksum, msg, repr(exp))
return None
def get_app_name(a, app_dir, is_apk):
def get_apk_name(app_dic):
"""Get app name."""
base = Path(app_dir)
if is_apk:
if a:
# Parsed Androguard APK Object
try:
app_name = a.get_app_name()
if app_name:
return app_name
except Exception:
logger.warning('Failed to get app name from parsed APK object')
# Look for app_name in values folder.
val = base / 'apktool_out' / 'res' / 'values'
if val.exists():
try:
return get_app_name_from_values_folder(val.as_posix())
except Exception:
logger.error('Failed to get app name from values folder')
real_name = ''
base = Path(app_dic['app_dir'])
# Check if it's an APK and try to retrieve the app name
if app_dic.get('androguard_apk_name') or app_dic.get('apk_features'):
app_name = (
app_dic.get('androguard_apk_name')
or app_dic.get('apk_features', {}).get('application_label')
)
if app_name:
real_name = app_name
else:
# Fallback: Look for app_name in the values folder
values_path = base / 'apktool_out' / 'res' / 'values'
if values_path.exists():
try:
real_name = get_app_name_from_values_folder(values_path.as_posix())
except Exception:
logger.error('Failed to get app name from values folder')
# Check if it's source code and try to retrieve the app name
else:
# For source code
try:
strings_path = base / 'app' / 'src' / 'main' / 'res' / 'values'
eclipse_path = base / 'res' / 'values'
if strings_path.exists():
return get_app_name_from_values_folder(
strings_path.as_posix())
elif eclipse_path.exists():
return get_app_name_from_values_folder(
eclipse_path.as_posix())
# Check paths for values folders
paths_to_check = [
base / 'app' / 'src' / 'main' / 'res' / 'values',
base / 'res' / 'values',
]
for path in paths_to_check:
if path.exists():
real_name = get_app_name_from_values_folder(path.as_posix())
break
except Exception:
logger.error('Failed to get app name')
logger.warning('Cannot find app name')
return ''
logger.error('Failed to get app name from source code')
if not real_name:
logger.warning('Cannot find app name')
# Update the app dictionary
app_dic['real_name'] = real_name
def get_app_name_from_values_folder(values_dir):
@@ -42,15 +42,20 @@ HASH_FUNCS = {
}
def get_hardcoded_cert_keystore(checksum, files):
def get_hardcoded_cert_keystore(app_dic):
"""Returns the hardcoded certificate keystore."""
app_dic['file_analysis'] = []
checksum = app_dic['md5']
try:
files = app_dic.get('files') or app_dic.get('apk_files')
msg = 'Getting Hardcoded Certificates/Keystores'
logger.info(msg)
append_scan_status(checksum, msg)
findings = []
certz = []
key_store = []
if not files:
return
for file_name in files:
if '.' not in file_name:
continue
@@ -66,7 +71,7 @@ def get_hardcoded_cert_keystore(checksum, files):
if key_store:
desc = 'Hardcoded Keystore found.'
findings.append({'finding': desc, 'files': key_store})
return findings
app_dic['file_analysis'] = findings
except Exception as exp:
msg = 'Getting Hardcoded Certificates/Keystores'
append_scan_status(checksum, msg, repr(exp))
@@ -183,26 +188,29 @@ def apksigtool_cert(checksum, apk_path, tools_dir):
av1 = True
else:
av1 = False
_, sig_block = extract_v2_sig(apk_path)
for pair in parse_apk_signing_block(sig_block).pairs:
b = pair.value
if isinstance(b, APKSignatureSchemeBlock):
signed = True
for signer in b.signers:
av2 = b.is_v2()
av3 = b.is_v3()
if b.is_v3():
min_sdk = signer.min_sdk
certs_no = len(signer.signed_data.certificates)
for cert in signer.signed_data.certificates:
d = get_cert_details(cert.data)
for i in d:
if i not in certs:
certs.append(i)
p = get_pub_key_details(signer.public_key.data)
for j in p:
if j not in pub_keys:
pub_keys.append(j)
try:
_, sig_block = extract_v2_sig(apk_path)
for pair in parse_apk_signing_block(sig_block).pairs:
b = pair.value
if isinstance(b, APKSignatureSchemeBlock):
signed = True
for signer in b.signers:
av2 = b.is_v2()
av3 = b.is_v3()
if b.is_v3():
min_sdk = signer.min_sdk
certs_no = len(signer.signed_data.certificates)
for cert in signer.signed_data.certificates:
d = get_cert_details(cert.data)
for i in d:
if i not in certs:
certs.append(i)
p = get_pub_key_details(signer.public_key.data)
for j in p:
if j not in pub_keys:
pub_keys.append(j)
except Exception:
logger.warning('Failed to get signature versions with apksigtool')
if signed:
certlist.append('Binary is signed')
@@ -289,12 +297,13 @@ def get_cert_data(checksum, a, app_path, tools_dir):
}
def cert_info(a, app_dic, man_dict):
def cert_info(app_dic, man_dict):
"""Return certificate information."""
try:
msg = 'Reading Code Signing Certificate'
logger.info(msg)
append_scan_status(app_dic['md5'], msg)
a = app_dic.get('androguard_apk')
manifestfile = None
manidat = ''
files = []
@@ -71,6 +71,17 @@ def permission_transform(perm_mappings):
def code_analysis(checksum, app_dir, typ, manifest_file, android_permissions):
"""Perform the code analysis."""
result = {
'api': {},
'behaviour': {},
'perm_mappings': {},
'findings': {},
'niap': {},
'urls_list': [],
'urls': [],
'emails': [],
'sbom': {},
}
try:
root = Path(settings.BASE_DIR) / 'StaticAnalyzer' / 'views'
and_rules = root / 'android' / 'rules'
@@ -78,15 +89,6 @@ def code_analysis(checksum, app_dir, typ, manifest_file, android_permissions):
api_rules = and_rules / 'android_apis.yaml'
perm_rules = and_rules / 'android_permissions.yaml'
niap_rules = and_rules / 'android_niap.yaml'
code_findings = {}
api_findings = {}
perm_mappings = {}
behaviour_findings = {}
niap_findings = {}
email_n_file = []
url_n_file = []
url_list = []
sbom = {}
app_dir = Path(app_dir)
src = get_android_src_dir(app_dir, typ).as_posix() + '/'
skp = settings.SKIP_CLASS_PATH
@@ -104,13 +106,14 @@ def code_analysis(checksum, app_dir, typ, manifest_file, android_permissions):
file_data = sast.read_files()
# SBOM Analysis
sbom = sbom_analysis.sbom(app_dir, file_data)
result['sbom'] = sbom_analysis.sbom(app_dir, file_data)
msg = 'Android SBOM Analysis Completed'
logger.info(msg)
append_scan_status(checksum, msg)
# Code Analysis
code_findings = sast.run_rules(file_data, code_rules.as_posix())
result['findings'] = sast.run_rules(
file_data, code_rules.as_posix())
msg = 'Android SAST Completed'
logger.info(msg)
append_scan_status(checksum, msg)
@@ -120,19 +123,21 @@ def code_analysis(checksum, app_dir, typ, manifest_file, android_permissions):
logger.info(msg)
append_scan_status(checksum, msg)
sast = SastEngine(options, src)
api_findings = sast.run_rules(file_data, api_rules.as_posix())
result['api'] = sast.run_rules(
file_data, api_rules.as_posix())
msg = 'Android API Analysis Completed'
logger.info(msg)
append_scan_status(checksum, msg)
# Permission Mapping
rule_file = get_perm_rules(checksum, perm_rules, android_permissions)
rule_file = get_perm_rules(
checksum, perm_rules, android_permissions)
if rule_file:
msg = 'Android Permission Mapping Started'
logger.info(msg)
append_scan_status(checksum, msg)
sast = SastEngine(options, src)
perm_mappings = permission_transform(
result['perm_mappings'] = permission_transform(
sast.run_rules(file_data, rule_file.name))
msg = 'Android Permission Mapping Completed'
logger.info(msg)
@@ -141,7 +146,7 @@ def code_analysis(checksum, app_dir, typ, manifest_file, android_permissions):
# Behavior Analysis
sast = SastEngine(options, src)
behaviour_findings = behaviour_analysis.analyze(
result['behaviour'] = behaviour_analysis.analyze(
checksum, sast, file_data)
# NIAP Scan
@@ -151,13 +156,14 @@ def code_analysis(checksum, app_dir, typ, manifest_file, android_permissions):
append_scan_status(checksum, msg)
niap_options = {
'choice_rules': niap_rules.as_posix(),
'alternative_path': manifest_file if manifest_file else '',
'alternative_path': str(manifest_file) if manifest_file else '',
'choice_extensions': {'.java', '.xml'},
'ignore_paths': skp,
}
cengine = ChoiceEngine(niap_options, src)
file_data = cengine.read_files()
niap_findings = cengine.run_rules(file_data, niap_rules.as_posix())
result['niap'] = cengine.run_rules(
file_data, niap_rules.as_posix())
msg = 'NIAP Analysis Completed'
logger.info(msg)
append_scan_status(checksum, msg)
@@ -182,25 +188,14 @@ def code_analysis(checksum, app_dir, typ, manifest_file, android_permissions):
relative_java_path = pfile.as_posix().replace(src, '')
urls, urls_nf, emails_nf = url_n_email_extract(
content, relative_java_path)
url_list.extend(urls)
url_n_file.extend(urls_nf)
email_n_file.extend(emails_nf)
result['urls_list'].extend(urls)
result['urls'].extend(urls_nf)
result['emails'].extend(emails_nf)
msg = 'Email and URL Extraction Completed'
logger.info(msg)
append_scan_status(checksum, msg)
code_an_dic = {
'api': api_findings,
'behaviour': behaviour_findings,
'perm_mappings': perm_mappings,
'findings': code_findings,
'niap': niap_findings,
'urls_list': url_list,
'urls': url_n_file,
'emails': email_n_file,
'sbom': sbom,
}
return code_an_dic
except Exception as exp:
msg = 'Failed to perform code analysis'
logger.exception(msg)
append_scan_status(checksum, msg, repr(exp))
return result
@@ -10,6 +10,7 @@ import subprocess
import threading
import stat
from pathlib import Path
from tempfile import gettempdir
from django.conf import settings
@@ -153,3 +154,37 @@ def apk_2_java(checksum, app_path, app_dir, dwd_tools_dir):
msg = 'Decompiling with JADX failed'
logger.exception(msg)
append_scan_status(checksum, msg, repr(exp))
def run_apktool(app_path, app_dir, tools_dir):
"""Get readable AndroidManifest.xml from APK."""
try:
if (len(settings.APKTOOL_BINARY) > 0
and Path(settings.APKTOOL_BINARY).exists()):
apktool_path = Path(settings.APKTOOL_BINARY)
else:
apktool_path = tools_dir / 'apktool_2.10.0.jar'
# Prepare output directory and manifest file paths
output_dir = app_dir / 'apktool_out'
# Run apktool to extract AndroidManifest.xml
args = [find_java_binary(),
'-jar',
'-Djdk.util.zip.disableZip64ExtraFieldValidation=true',
str(apktool_path),
'--match-original',
'--frame-path',
gettempdir(),
'-f', '-s', 'd',
str(app_path),
'-o',
str(output_dir)]
logger.info('Converting AXML to XML with apktool')
with open(os.devnull, 'w') as fnull:
subprocess.run(
args,
stdout=fnull,
stderr=subprocess.STDOUT,
timeout=settings.JADX_TIMEOUT)
except Exception:
logger.warning('apktool failed to extract AndroidManifest.xml')
@@ -144,7 +144,7 @@ def get_context_from_analysis(app_dic,
'manifest_analysis': manifest_analysis,
'network_security': man_an_dic['network_security'],
'binary_analysis': bin_anal,
'file_analysis': app_dic['certz'],
'file_analysis': app_dic['file_analysis'],
'android_api': code_an_dic['api'],
'code_analysis': code,
'niap_analysis': code_an_dic['niap'],
@@ -210,7 +210,7 @@ def save_or_update(update_type,
'MALWARE_PERMISSIONS': man_an_dic['malware_permissions'],
'MANIFEST_ANALYSIS': man_an_dic['manifest_anal'],
'BINARY_ANALYSIS': bin_anal,
'FILE_ANALYSIS': app_dic['certz'],
'FILE_ANALYSIS': app_dic['file_analysis'],
'ANDROID_API': code_an_dic['api'],
'CODE_ANALYSIS': code_an_dic['findings'],
'NIAP_ANALYSIS': code_an_dic['niap'],
@@ -5,10 +5,11 @@ import fnmatch
import logging
import os
from shutil import copy2, copytree
from xml.dom import minidom
from pathlib import Path
import subprocess
from defusedxml.minidom import parseString
from lxml import etree
from django.conf import settings
@@ -22,6 +23,9 @@ from mobsf.MobSF.utils import (
from mobsf.StaticAnalyzer.tools.androguard4 import (
axml,
)
from mobsf.StaticAnalyzer.views.common.shared_func import (
RESERVED_FILE_NAMES,
)
logger = logging.getLogger(__name__)
@@ -145,39 +149,47 @@ def find_icon_path_zip(checksum, res_dir, icon_paths_from_manifest):
msg = 'Failed to find icon path'
logger.exception(msg)
append_scan_status(checksum, msg, repr(exp))
# PNG icon lookup functions above ^
# SVG/XML icon lookup functions below
def get_icon_src(a, app_dic, res_dir):
"""
Returns a dict with isHidden boolean and a relative path.
path is a full path (not relative to resource folder)
"""
def get_icon_apk_res(app_dic):
"""Get icon path from APK resource."""
icon_src = ''
checksum = app_dic['md5']
try:
msg = 'Fetching icon path'
logger.info(msg)
append_scan_status(app_dic['md5'], msg)
icon_src = ''
append_scan_status(checksum, msg)
app_dir = Path(app_dic['app_dir'])
icon_resolution = 0xFFFE - 1
res_path = app_dir / 'res'
apktool_res_path = app_dir / 'apktool_out' / 'res'
icon_name = None
if a:
# If icon is found in androguard or aapt2
icon_name = app_dic.get('androguard_apk_icon') or app_dic.get(
'apk_features', {}).get('application_icon')
if is_path_traversal(icon_name):
logger.warning('Path traversal detected in icon path')
icon_name = None
# Handle reserved file names case
if icon_name and any(
icon_name.startswith(x) for x in RESERVED_FILE_NAMES):
icon_name = str(Path('_conflict_') / icon_name)
# androguard/aapt2 cannot find icon file, fallback to res or apktool.
if not res_path.exists() and apktool_res_path.exists() and not icon_name:
logger.warning('Cannot find res directory,'
' fallback to apktool res directory')
try:
icon_name = a.get_app_icon(max_dpi=icon_resolution)
if icon_name and is_path_traversal(icon_name):
icon_name = None
copytree(apktool_res_path, res_path, dirs_exist_ok=True)
except Exception:
logger.warning('Failed to get icon from parsed APK object')
icon_name = None
if not icon_name:
# androguard cannot find icon file.
pass
if res_path.exists() and not icon_name:
icon_name = ''
logger.warning('androguard cannot find icon resource')
icon_name = guess_icon_path(res_dir)
logger.warning('androguard/aapt2 cannot find icon resource')
icon_name = guess_icon_path(str(res_path))
icon_src = icon_name
if icon_name.endswith('.xml'):
if icon_name and icon_name.endswith('.xml'):
apktool_res = False
# Can be vector XML/XML pointing to vector files
# Convert AXML to XML for vector
@@ -198,16 +210,16 @@ def get_icon_src(a, app_dic, res_dir):
icon_path = ipath.as_posix()
else:
# When icon xml point to other vector files
icon_path = get_icon_svg_from_xml(
app_dir, icon_name)
icon_path = get_icon_svg_from_xml(app_dir, icon_name)
if icon_path:
icon_src = icon_path
else:
# if we cannot find from xml
icon_src = guess_icon_path(res_dir)
icon_src = guess_icon_path(str(res_path))
else:
# We found png icon, the easy path
icon_src = (app_dir / icon_name).as_posix()
if icon_src.endswith('.xml'):
logger.warning('Cannot find icon file from xml')
icon_src = ''
@@ -221,40 +233,30 @@ def get_icon_src(a, app_dic, res_dir):
except Exception as exp:
msg = 'Failed to fetch icon path'
logger.exception(msg)
append_scan_status(app_dic['md5'], msg, repr(exp))
append_scan_status(checksum, msg, repr(exp))
return icon_src
def get_icon_apk(apk, app_dic):
def get_icon_apk(app_dic):
"""Get/Guess icon from APK binary."""
app_dir = Path(app_dic['app_dir'])
icon_file = ''
res_path = app_dir / 'res'
if not res_path.exists():
logger.warning('Cannot find res directory,'
' using apktool res directory')
# If res directory is not found or named differently
# piggyback on apktool decompiled resources
try:
apk_tool_res = app_dir / 'apktool_out' / 'res'
copytree(apk_tool_res, res_path, dirs_exist_ok=True)
except Exception:
pass
if res_path.exists():
app_dic['icon_path'] = ''
try:
# Icon lookup in res directory
icon_file = get_icon_src(
apk,
app_dic,
res_path.as_posix())
icon_file = get_icon_apk_res(app_dic)
if icon_file:
src = Path(icon_file)
# Copy PNG/SVG to Downloads
icon = app_dic['md5'] + '-icon' + src.suffix.lower()
out = Path(settings.DWD_DIR) / icon
if src and src.exists() and src.is_file():
copy2(src.as_posix(), out.as_posix())
app_dic['icon_path'] = out.name
except Exception:
logger.exception('Failed to get icon from APK')
if icon_file:
src = Path(icon_file)
# Copy PNG/SVG to Downloads
icon = app_dic['md5'] + '-icon' + src.suffix.lower()
out = Path(settings.DWD_DIR) / icon
if src and src.exists() and src.is_file():
copy2(src.as_posix(), out.as_posix())
app_dic['icon_path'] = out.name
# PNG icon lookup functions above ^
# SVG/XML icon lookup functions below
def transform_svg(fpath, bpath, output):
@@ -280,8 +282,7 @@ def get_icon_svg_from_xml(app_dir, icon_xml_file):
"""
try:
icon_xml = app_dir / 'apktool_out' / icon_xml_file
parsed = minidom.parseString(
icon_xml.read_text('utf8', 'ignore'))
parsed = parseString(icon_xml.read_text('utf8', 'ignore'))
foreground = parsed.getElementsByTagName('foreground')
background = parsed.getElementsByTagName('background')
ficon = foreground[0].getAttribute(
+11 -46
View File
@@ -26,15 +26,12 @@ from mobsf.StaticAnalyzer.views.common.firebase import (
from mobsf.StaticAnalyzer.views.common.appsec import (
get_android_dashboard,
)
from mobsf.StaticAnalyzer.views.android.app import (
parse_apk,
)
from mobsf.StaticAnalyzer.views.android.manifest_analysis import (
manifest_analysis,
)
from mobsf.StaticAnalyzer.views.android.manifest_utils import (
get_manifest,
manifest_data,
extract_manifest_data,
get_parsed_manifest,
)
from mobsf.StaticAnalyzer.views.android.strings import (
get_strings_metadata,
@@ -87,6 +84,7 @@ def common_analysis(request, app_dic, rescan, api, analysis_type):
app_dic['sha1'], app_dic['sha256'] = hash_gen(
checksum,
app_dic['app_path'])
app_dic['zipped'] = analysis_type
app_dic['files'] = unzip(
checksum,
app_dic['app_path'],
@@ -97,53 +95,24 @@ def common_analysis(request, app_dic, rescan, api, analysis_type):
request,
f'{analysis_type.upper()} file is invalid or corrupt',
api)
app_dic['certz'] = get_hardcoded_cert_keystore(
checksum,
app_dic['files'])
get_hardcoded_cert_keystore(app_dic)
app_dic['playstore'] = {'error': True}
# Parse APK with Androguard4
apk = parse_apk(
checksum,
app_dic['app_path'])
if analysis_type == 'aar':
# AAR has manifest and sometimes certificate
mani_file, ns, mani_xml = get_manifest(
checksum,
app_dic['app_path'],
app_dic['app_dir'],
app_dic['tools_dir'],
'aar',
apk,
)
app_dic['manifest_file'] = mani_file
app_dic['ns'] = ns
app_dic['parsed_xml'] = mani_xml
man_data_dic = manifest_data(
checksum,
app_dic['parsed_xml'],
ns)
man_an_dic = manifest_analysis(
checksum,
app_dic['parsed_xml'],
ns,
man_data_dic,
'',
app_dic['app_dir'],
)
get_parsed_manifest(app_dic)
man_data_dic = extract_manifest_data(app_dic)
man_an_dic = manifest_analysis(app_dic, man_data_dic)
# Malware Permission check
mal_perms = permissions.check_malware_permission(
checksum,
man_data_dic['perm'])
man_an_dic['malware_permissions'] = mal_perms
cert_dic = cert_info(
apk,
app_dic,
man_data_dic)
cert_dic = cert_info(app_dic, man_data_dic)
else:
app_dic['manifest_file'] = None
app_dic['parsed_xml'] = ''
app_dic['manifest_parsed_xml'] = None
app_dic['manifest_namespace'] = None
man_data_dic = {
'services': [],
'activities': [],
@@ -210,11 +179,8 @@ def common_analysis(request, app_dic, rescan, api, analysis_type):
code_an_dic)
# Get the strings and metadata
get_strings_metadata(
checksum,
apk,
app_dic['app_dir'],
app_dic,
elf_dict['elf_strings'],
APK_TYPE,
['.java'],
code_an_dic)
# Firebase DB Check
@@ -226,7 +192,6 @@ def common_analysis(request, app_dic, rescan, api, analysis_type):
checksum,
code_an_dic['urls_list'])
app_dic['zipped'] = analysis_type
context = save_get_ctx(
app_dic,
man_data_dic,
@@ -60,6 +60,8 @@ ANDROID_API_LEVEL_MAP = {
'32': '12L',
'33': '13',
'34': '14',
'35': '15',
'36': '16',
}
@@ -181,9 +183,14 @@ def get_browsable_activities(node, ns):
logger.exception('Getting Browsable Activities')
def manifest_analysis(checksum, mfxml, ns, man_data_dic, src_type, app_dir):
def manifest_analysis(app_dic, man_data_dic):
"""Analyse manifest file."""
# pylint: disable=C0301
checksum = app_dic['md5']
mfxml = app_dic['manifest_parsed_xml']
ns = app_dic['manifest_namespace']
src_type = app_dic['zipped']
app_dir = app_dic['app_dir']
try:
msg = 'Manifest Analysis Started'
logger.info(msg)
@@ -1,22 +1,18 @@
# -*- coding: utf_8 -*-
"""Android manifest analysis utils."""
import logging
import os
import re
import subprocess
import tempfile
from pathlib import Path
from xml.dom import minidom
from xml.parsers.expat import ExpatError
from defusedxml.minidom import parseString
from bs4 import BeautifulSoup
from django.conf import settings
from mobsf.MobSF.utils import (
append_scan_status,
find_java_binary,
is_file_exists,
)
from mobsf.StaticAnalyzer.views.android.converter import (
run_apktool,
)
# pylint: disable=E0401
@@ -31,97 +27,79 @@ ANDROID_8_0_LEVEL = 26
ANDROID_MANIFEST_FILE = 'AndroidManifest.xml'
def get_manifest_file(app_dir, app_path, tools_dir, typ, apk):
"""Read the manifest file."""
def get_manifest_file(app_dic):
"""Get AndroidManifest.xml file path.
Used by get_parsed_manifest() and manifest_view.run()
"""
manifest = None
try:
manifest = ''
app_path = Path(app_dic['app_path'])
app_dir = Path(app_dic['app_dir'])
tools_dir = Path(app_dic['tools_dir'])
typ = app_dic['zipped']
checksum = app_dic['md5']
androguard_xml = app_dic.get('androguard_manifest_xml')
if typ == 'aar':
logger.info('Getting AndroidManifest.xml from AAR')
manifest = os.path.join(app_dir, ANDROID_MANIFEST_FILE)
manifest = app_dir / ANDROID_MANIFEST_FILE
elif typ == 'apk':
logger.info('Getting AndroidManifest.xml from APK')
manifest = get_manifest_apk(app_path, app_dir, tools_dir, apk)
manifest = app_dir / 'apktool_out' / ANDROID_MANIFEST_FILE
if manifest.exists():
return manifest
# Run apktool to extract AndroidManifest.xml
manifest.parent.mkdir(parents=True, exist_ok=True)
run_apktool(app_path, app_dir, tools_dir)
if not manifest.exists() and androguard_xml:
logger.warning(
'apktool failed to extract AndroidManifest.xml,'
' fallback to androguard')
manifest.write_bytes(androguard_xml)
elif not androguard_xml:
msg = ('Failed to extract AndroidManifest.xml'
' from APK with apktool and androguard')
logger.error(msg)
append_scan_status(checksum, msg, 'apktool and androguard failed')
elif typ == 'eclipse':
logger.info('Getting AndroidManifest.xml'
' from Eclipse project source code')
manifest = app_dir / ANDROID_MANIFEST_FILE
elif typ == 'studio':
logger.info('Getting AndroidManifest.xml'
' from Android Studio project source code')
manifest = app_dir / 'app' / 'src' / \
'main' / ANDROID_MANIFEST_FILE
else:
logger.info('Getting AndroidManifest.xml from Source Code')
if typ == 'eclipse':
manifest = os.path.join(app_dir, ANDROID_MANIFEST_FILE)
elif typ == 'studio':
manifest = os.path.join(
app_dir,
f'app/src/main/{ANDROID_MANIFEST_FILE}')
return manifest
logger.error('Unknown project type')
except Exception:
logger.exception('Getting AndroidManifest.xml file')
def get_android_manifest_androguard(apk, app_dir):
"""Get AndroidManifest.xml using Androguard4."""
try:
logger.info('Extracting AndroidManifest.xml with Androguard')
if not apk:
logger.warning('Androgaurd APK parsing failed')
return
manifest = apk.get_android_manifest_axml()
if not manifest:
return
manifest_file = Path(app_dir) / 'apktool_out' / ANDROID_MANIFEST_FILE
manifest_file.write_bytes(manifest.get_xml())
except Exception:
logger.exception('Error Extracting AndroidManifest.xml with Androguard')
return None
def get_manifest_apk(app_path, app_dir, tools_dir, apk):
"""Get readable AndroidManifest.xml.
Should be called before get_icon_apk() function
"""
try:
manifest = None
if (len(settings.APKTOOL_BINARY) > 0
and is_file_exists(settings.APKTOOL_BINARY)):
apktool_path = settings.APKTOOL_BINARY
else:
apktool_path = os.path.join(tools_dir, 'apktool_2.10.0.jar')
output_dir = os.path.join(app_dir, 'apktool_out')
args = [find_java_binary(),
'-jar',
'-Djdk.util.zip.disableZip64ExtraFieldValidation=true',
apktool_path,
'--match-original',
'--frame-path',
tempfile.gettempdir(),
'-f', '-s', 'd',
app_path,
'-o',
output_dir]
manifest = os.path.join(output_dir, ANDROID_MANIFEST_FILE)
if is_file_exists(manifest):
# APKTool already created readable XML
return manifest
logger.info('Converting AXML to XML')
subprocess.check_output(args)
except subprocess.CalledProcessError:
# APK tool failed
logger.warning('apktool failed to extract AndroidManifest.xml')
get_android_manifest_androguard(apk, app_dir)
except Exception:
logger.exception('Getting Manifest file')
return manifest
def get_xml_namespace(xml_str):
"""Get namespace."""
m = re.search(r'manifest (.{1,250}?):', xml_str)
if m:
return m.group(1)
logger.warning('XML namespace not found')
return None
match = re.search(r'manifest (.{1,250}?):', xml_str)
if not match:
logger.warning('XML namespace not found')
return None
namespace = match.group(1)
# Handle standard and non-standard namespaces
if namespace == 'xmlns':
namespace = 'android'
elif namespace != 'android':
logger.warning('Non-standard XML namespace: %s', namespace)
return namespace
def get_fallback():
logger.warning('Using Fake XML to continue the Analysis')
return minidom.parseString(
return parseString(
(r'<?xml version="1.0" encoding="utf-8"?><manifest xmlns:android='
r'"http://schemas.android.com/apk/res/android" '
r'android:versionCode="Failed" '
@@ -141,46 +119,63 @@ def bs4_xml_parser(xml_str):
return None
def get_manifest(checksum, app_path, app_dir, tools_dir, typ, apk):
"""Get the manifest file."""
def get_parsed_manifest(app_dic):
"""Get the parsed manifest XML, file path and namespace."""
checksum = app_dic['md5']
parsed_xml = None
manifest_file = None
app_dic['manifest_file'] = None
app_dic['manifest_namespace'] = 'android'
app_dic['manifest_parsed_xml'] = get_fallback()
try:
ns = 'android'
manifest_file = get_manifest_file(
app_dir,
app_path,
tools_dir,
typ,
apk)
mfile = Path(manifest_file)
if not mfile.exists():
logger.warning('apktool failed to extract '
'AndroidManifest.xml')
return manifest_file, ns, get_fallback()
manifest_file = get_manifest_file(app_dic)
if not (manifest_file and manifest_file.exists()):
logger.warning('APK AndroidManifest.xml file not found')
app_dic['manifest_file'] = manifest_file
return
app_dic['manifest_file'] = manifest_file
msg = 'Parsing AndroidManifest.xml'
logger.info(msg)
append_scan_status(checksum, msg)
xml_str = mfile.read_text('utf-8', 'ignore')
ns = get_xml_namespace(xml_str)
if ns and ns == 'xmlns':
ns = 'android'
if ns and ns != 'android':
logger.warning('Non standard XML namespace: %s', ns)
# Parse manifest XML
xml_str = manifest_file.read_text('utf-8', 'ignore')
app_dic['manifest_namespace'] = get_xml_namespace(xml_str)
# apktool generated AndroidManifest.xml for APK
parsed_xml = parseString(xml_str)
app_dic['manifest_parsed_xml'] = parsed_xml
return
except Exception:
try:
return manifest_file, ns, minidom.parseString(xml_str)
except ExpatError:
logger.warning('Parsing AndroidManifest.xml failed')
return manifest_file, ns, minidom.parseString(
bs4_xml_parser(xml_str))
except Exception as exp:
msg = 'Parsing AndroidManifest.xml failed'
logger.exception(msg)
append_scan_status(checksum, msg, repr(exp))
return manifest_file, ns, get_fallback()
logger.warning('Failed parsing AndroidManifest.xml, fallback to androguard')
parsed_xml = parseString(app_dic['androguard_manifest_xml'])
# Overwrite the file with androguard generated XML
manifest_file.write_bytes(app_dic['androguard_manifest_xml'])
app_dic['manifest_parsed_xml'] = parsed_xml
return
except Exception:
logger.warning('Failed parsing androguard AndroidManifest.xml'
', fallback to bs4')
try:
parsed_xml = parseString(bs4_xml_parser(xml_str))
app_dic['manifest_parsed_xml'] = parsed_xml
return
except Exception as exp:
msg = 'Parsing AndroidManifest.xml using all methods'
logger.exception(msg)
append_scan_status(checksum, msg, repr(exp))
def manifest_data(checksum, mfxml, ns):
"""Extract manifest data."""
def extract_manifest_data(app_dic):
"""Extract manifest data.
Data from apk_features (aapt2) is also available as a fallback.
"""
checksum = app_dic['md5']
try:
mfxml = app_dic['manifest_parsed_xml']
ns = app_dic['manifest_namespace']
msg = 'Extracting Manifest Data'
logger.info(msg)
append_scan_status(checksum, msg)
@@ -219,12 +214,12 @@ def manifest_data(checksum, mfxml, ns):
# Esteve 08.08.2016 - begin - If android:targetSdkVersion
# is not set, the default value is the one of the
# minSdkVersiontargetsdk
# = node.getAttribute (f'{ns}:targetSdkVersion')
if node.getAttribute(f'{ns}:targetSdkVersion'):
if app_dic.get('apk_features', {}).get('target_sdk_version'):
targetsdk = app_dic['apk_features']['target_sdk_version']
elif node.getAttribute(f'{ns}:targetSdkVersion'):
targetsdk = node.getAttribute(f'{ns}:targetSdkVersion')
else:
targetsdk = node.getAttribute(f'{ns}:minSdkVersion')
# End
for node in manifest:
package = node.getAttribute('package')
androidversioncode = node.getAttribute(f'{ns}:versionCode')
@@ -277,6 +272,8 @@ def manifest_data(checksum, mfxml, ns):
android_permission_tags = ('com.google.', 'android.', 'com.google.')
for permission in permissions:
perm.append(permission.getAttribute(f'{ns}:name'))
if not perm and app_dic.get('apk_features', {}).get('permissions'):
perm = app_dic['apk_features']['permissions']
for full_perm in perm:
# For general android permissions
prm = full_perm
@@ -299,7 +296,15 @@ def manifest_data(checksum, mfxml, ns):
'Unknown permission',
'Unknown permission from android reference',
]
pkg_backup = app_dic.get('apk_features', {}).get('package')
mainact_backup = app_dic.get('apk_features', {}).get('launchable_activity')
minsdk_backup = app_dic.get('apk_features', {}).get('min_sdk_version')
if not package and pkg_backup:
package = pkg_backup
if not mainact and mainact_backup:
mainact = mainact_backup
if not minsdk and minsdk_backup:
minsdk = minsdk_backup
man_data_dic = {
'services': svc,
'activities': act,
@@ -1,9 +1,10 @@
# -*- coding: utf_8 -*-
"""Module for network security analysis."""
import logging
from xml.dom import minidom
from pathlib import Path
from defusedxml.minidom import parseString
from mobsf.MobSF.utils import (
append_scan_status,
is_path_traversal,
@@ -73,7 +74,7 @@ def analysis(checksum, app_dir, config, is_debuggable, src_type):
msg = 'Parsing Network Security config'
logger.info(msg)
append_scan_status(checksum, msg)
parsed = minidom.parseString(netsec_conf)
parsed = parseString(netsec_conf)
finds = []
summary = {HIGH: 0, WARNING: 0, INFO: 0, SECURE: 0}
# Base Config
@@ -17,9 +17,21 @@ from mobsf.MobSF.utils import (
logger = logging.getLogger(__name__)
def get_app_details(checksum, package_id):
def get_app_details(app_dic, man_data):
"""Get App Details form PlayStore."""
checksum = app_dic['md5']
app_dic['playstore'] = {
'error': True,
'description': 'Failed to identify the package name',
}
try:
if man_data.get('packagename'):
package_id = man_data['packagename']
elif app_dic.get('apk_features', {}).get('package'):
package_id = app_dic['apk_features']['package']
else:
logger.warning('Package Name not found')
return
msg = f'Fetching Details from Play Store: {package_id}'
logger.info(msg)
append_scan_status(checksum, msg)
@@ -33,7 +45,7 @@ def get_app_details(checksum, package_id):
det['androidVersionText'] = ''
except Exception:
det = app_search(checksum, package_id)
return det
app_dic['playstore'] = det
def app_search(checksum, app_id):
+6 -8
View File
@@ -60,11 +60,13 @@ def so_analysis(request, app_dic, rescan, api):
app_dic['sha1'], app_dic['sha256'] = hash_gen(
checksum,
app_dic['app_path'])
app_dic['zipped'] = 'so'
app_dic['files'] = []
app_dic['certz'] = []
app_dic['file_analysis'] = []
app_dic['playstore'] = {'error': True}
app_dic['manifest_file'] = None
app_dic['parsed_xml'] = ''
app_dic['manifest_namespace'] = None
app_dic['manifest_parsed_xml'] = None
app_dic['mani'] = ''
man_data_dic = {
'services': [],
@@ -111,7 +113,7 @@ def so_analysis(request, app_dic, rescan, api):
app_dic['app_dir'],
'elf')
# File Analysis is used to store symbols from so
app_dic['certz'] = get_symbols(
app_dic['file_analysis'] = get_symbols(
elf_dict['elf_symbols'])
apkid_results = {}
code_an_dic = {
@@ -127,12 +129,9 @@ def so_analysis(request, app_dic, rescan, api):
}
# Get the strings and metadata from shared object
get_strings_metadata(
checksum,
None,
None,
app_dic,
elf_dict['elf_strings'],
None,
None,
code_an_dic)
# Firebase DB Check
code_an_dic['firebase'] = firebase_analysis(
@@ -149,7 +148,6 @@ def so_analysis(request, app_dic, rescan, api):
app_dic['tools_dir'])
trackers = trk.get_trackers_domains_or_deps(
code_an_dic['domains'], [])
app_dic['zipped'] = 'so'
context = save_get_ctx(
app_dic,
man_data_dic,
+38 -30
View File
@@ -50,7 +50,7 @@ def strings_from_so(checksum, elf_strings):
return sos
def strings_from_apk(checksum, apk):
def strings_from_apk(checksum, app_dic):
"""Extract Strings from an APK."""
results = {
'strings': [],
@@ -65,36 +65,42 @@ def strings_from_apk(checksum, apk):
logger.info(msg)
append_scan_status(checksum, msg)
rsrc = apk.get_android_resources()
if not rsrc:
return results
rsrc = app_dic.get('androguard_apk_resources')
if rsrc:
pkg = rsrc.get_packages_names()[0]
rsrc.get_strings_resources()
pkg = rsrc.get_packages_names()[0]
rsrc.get_strings_resources()
# Iterate over resource strings
for _res_key, res_value in rsrc.values.get(pkg, {}).items():
res_string = res_value.get('string')
if not res_string:
continue
for key, value in res_string:
if not value:
# Iterate over resource strings
for _res_key, res_value in rsrc.values.get(pkg, {}).items():
res_string = res_value.get('string')
if not res_string:
continue
# Extract Firebase credentials
if key == 'google_api_key' and GOOGLE_API_KEY_REGEX.match(value):
results['firebase_creds']['google_api_key'] = value
elif key == 'google_app_id' and GOOGLE_APP_ID_REGEX.match(value):
results['firebase_creds']['google_app_id'] = value
for key, value in res_string:
if not value:
continue
# Format and collect strings
formatted_str = f'"{key}" : "{value}"'
results['strings'].append(formatted_str)
# Extract Firebase credentials
if key == 'google_api_key' and GOOGLE_API_KEY_REGEX.match(value):
results['firebase_creds']['google_api_key'] = value
elif key == 'google_app_id' and GOOGLE_APP_ID_REGEX.match(value):
results['firebase_creds']['google_app_id'] = value
# Check for possible secrets
if is_secret_key(key) and ' ' not in value:
results['secrets'].append(formatted_str)
# Format and collect strings
formatted_str = f'"{key}" : "{value}"'
results['strings'].append(formatted_str)
# Check for possible secrets
if is_secret_key(key) and ' ' not in value:
results['secrets'].append(formatted_str)
elif app_dic.get('apk_strings'):
# No secret key check for APK strings
results['strings'] = app_dic['apk_strings']
else:
msg = 'Failed to extract String data from APK'
logger.warning(msg)
append_scan_status(checksum, msg)
return results
# Extract URLs and Emails from collected strings
results['strings'] = list(set(results['strings']))
@@ -129,9 +135,11 @@ def strings_from_code(checksum, src_dir, typ, exts):
return data
def get_strings_metadata(
checksum, apk, app_dir, elf_strings, typ, exts, code_dic):
def get_strings_metadata(app_dic, elf_strings, exts, code_dic):
"""Get Strings, secrets, entropies, URLs, emails."""
checksum = app_dic['md5']
typ = app_dic['zipped']
app_dir = app_dic['app_dir']
strings = {
'strings_apk_res': {},
'strings_so': [],
@@ -141,9 +149,9 @@ def get_strings_metadata(
urls_n_files = []
emails_n_files = []
secrets = []
if apk:
if app_dic.get('androguard_string_resources') or app_dic.get('apk_strings'):
# APK
apk_res = strings_from_apk(checksum, apk)
apk_res = strings_from_apk(checksum, app_dic)
code_dic['firebase_creds'] = apk_res['firebase_creds']
strings['strings_apk_res'] = apk_res['strings']
urls_list.extend(apk_res['urls_list'])
@@ -2,7 +2,6 @@
"""Module for manifest_view."""
import logging
import os
from pathlib import Path
from django.conf import settings
@@ -27,23 +26,19 @@ def run(request, checksum):
"""View the manifest."""
try:
supported = ['eclipse', 'studio', 'apk', 'aar']
directory = settings.BASE_DIR # BASE DIR
typ = request.GET['type'] # APK or SOURCE
if is_md5(checksum) and (typ in supported):
app_dir = os.path.join(
settings.UPLD_DIR, checksum + '/') # APP DIRECTORY
tools_dir = os.path.join(
directory, 'StaticAnalyzer/tools/') # TOOLS DIR
app_path = os.path.join(app_dir, checksum + '.apk')
manifest_file = get_manifest_file(
app_dir,
app_path,
tools_dir,
typ,
None)
mfile = Path(manifest_file)
if mfile.exists():
manifest = mfile.read_text('utf-8', 'ignore')
app_dir = Path(settings.UPLD_DIR) / checksum
app_dic = {
'md5': checksum,
'app_dir': app_dir,
'app_path': app_dir / f'{checksum}.apk',
'tools_dir': Path(settings.BASE_DIR) / 'StaticAnalyzer' / 'tools',
'zipped': typ,
}
manifest_file = get_manifest_file(app_dic)
if manifest_file and manifest_file.exists():
manifest = manifest_file.read_text('utf-8', 'ignore')
else:
manifest = ''
context = {
+1 -1
View File
@@ -71,7 +71,7 @@ def common_fields(findings, data):
# File Analysis
cert_files = None
cfp = []
for fa in data['file_analysis']:
for fa in data.get('file_analysis', []):
if isinstance(fa, str):
# FA is being used by so/dylib
continue
@@ -108,6 +108,13 @@ def async_analysis(checksum, api, file_name, func, *args, **kwargs):
return HttpResponseRedirect('/tasks')
def enqueued_task_init(checksum):
"""Store Task start."""
EnqueuedTask.objects.filter(checksum=checksum).update(
started_at=timezone.now(),
)
def update_enqueued_task(checksum, app_name, status):
"""Update the Enqueued Task and others that matches the checksum."""
EnqueuedTask.objects.filter(checksum=checksum).update(
@@ -124,7 +131,7 @@ def get_live_status(enq):
return enq.status
logs = get_scan_logs(enq.checksum)
if logs:
return logs[-1]
return logs[-1].get('status', '')
return enq.status
@@ -142,6 +149,7 @@ def list_tasks(request, api=False):
'app_name': enq.app_name,
'checksum': enq.checksum,
'created_at': enq.created_at,
'started_at': enq.started_at,
'completed_at': enq.completed_at,
'status': get_live_status(enq),
})
+114 -28
View File
@@ -20,6 +20,9 @@ from django.utils.html import escape
from django.http import HttpResponseRedirect
from mobsf.MobSF import settings
from mobsf.MobSF.security import (
sanitize_for_logging,
)
from mobsf.MobSF.utils import (
EMAIL_REGEX,
STRINGS_REGEX,
@@ -51,6 +54,16 @@ from mobsf.MobSF.views.authorization import (
logger = logging.getLogger(__name__)
RESERVED_FILE_NAMES = [
'AndroidManifest.xml',
'resources.arsc',
'META-INF/MANIFEST.MF',
'META-INF/CERT.SF',
'META-INF/CERT.RSA',
'META-INF/CERT.DSA',
'classes.dex']
for i in range(2, 50):
RESERVED_FILE_NAMES.append(f'classes{i}.dex')
def hash_gen(checksum, app_path) -> tuple:
@@ -77,49 +90,122 @@ def hash_gen(checksum, app_path) -> tuple:
append_scan_status(checksum, msg, repr(exp))
def is_reserved_file_conflict(file_path):
"""Check for reserved file conflict."""
if any(file_path.startswith(i) and file_path != i for i in RESERVED_FILE_NAMES):
return True
return False
def unzip(checksum, app_path, ext_path):
"""Unzip APK.
Unzip a APK archive while handling encrypted files, reserved file conflicts,
path traversal (Zip Slip), and permission adjustments. Some of the anti-analysis
techniques used by malware authors and packers are handled here.
Args:
checksum (str): The checksum of the file.
app_path (str): Path to the ZIP archive.
ext_path (str): Path to extract the files.
Returns:
list: A list of files extracted or an empty list if an error occurs.
"""
msg = 'Unzipping'
logger.info(msg)
append_scan_status(checksum, msg)
files = []
original_ext_path = ext_path
try:
files = []
with zipfile.ZipFile(app_path, 'r') as zipptr:
files = zipptr.namelist()
for fileinfo in zipptr.infolist():
filename = fileinfo.filename
if not isinstance(filename, str):
filename = str(
filename, encoding='utf-8', errors='replace')
files.append(filename)
zipptr.extract(filename, ext_path)
return files
ext_path = original_ext_path
# Skip encrypted files
if fileinfo.flag_bits & 0x1:
msg = ('Skipping encrypted file '
f'{sanitize_for_logging(fileinfo.filename)}')
logger.warning(msg)
continue
file_path = fileinfo.filename.rstrip('/\\') # Remove trailing slashes
# Decode the filename
if not isinstance(file_path, str):
file_path = file_path.decode('utf-8', errors='replace')
# Check for reserved file conflict
if is_reserved_file_conflict(file_path):
ext_path = str(Path(ext_path) / '_conflict_')
# Handle Zip Slip
if is_path_traversal(file_path):
msg = ('Zip slip detected. skipped extracting'
f' {sanitize_for_logging(file_path)}')
logger.error(msg)
continue
# Fix permissions
if fileinfo.is_dir():
# Directories should have rwxr-xr-x (755)
# Skip creating directories
continue
else:
# Files should have rw-r--r-- (644)
fileinfo.external_attr = (0o100644 << 16) | (
fileinfo.external_attr & 0xFFFF)
# Extract the file
try:
zipptr.extract(file_path, ext_path)
except Exception:
logger.warning(
'Failed to extract %s', sanitize_for_logging(file_path))
except Exception as exp:
msg = f'Unzipping Error - {str(exp)}'
logger.error(msg)
append_scan_status(checksum, msg, repr(exp))
# Fallback to OS unzip
ofiles = os_unzip(checksum, app_path, ext_path)
if not files:
files = ofiles
return files
def os_unzip(checksum, app_path, ext_path):
"""Unzip using OS utility."""
msg = 'Attempting to unzip with OS unzip utility'
logger.info(msg)
append_scan_status(checksum, msg)
try:
if platform.system() == 'Windows':
msg = 'Unzipping Error. Not yet implemented in Windows'
logger.warning(msg)
append_scan_status(checksum, msg)
else:
msg = 'Attempting to unzip with OS unzip utility'
logger.info(msg)
return []
unzip_b = shutil.which('unzip')
if not unzip_b:
msg = 'OS Unzip utility not found'
logger.warning(msg)
append_scan_status(checksum, msg)
try:
unzip_b = shutil.which('unzip')
subprocess.call(
[unzip_b, '-o', '-q', app_path, '-d', ext_path])
# Set permissions, packed files
# may not have proper permissions
set_permissions(ext_path)
dat = subprocess.check_output([unzip_b, '-qq', '-l', app_path])
dat = dat.decode('utf-8').split('\n')
files_det = ['Length Date Time Name']
files_det = files_det + dat
return files_det
except Exception as exp:
msg = 'Unzipping Error with OS unzip utility'
logger.exception(msg)
append_scan_status(checksum, msg, repr(exp))
return []
subprocess.call(
[unzip_b, '-o', '-q', app_path, '-d', ext_path])
# Set permissions, packed files
# may not have proper permissions
set_permissions(ext_path)
# List files in the unzipped directory
dat = subprocess.check_output([unzip_b, '-qq', '-l', app_path])
dat = dat.decode('utf-8').split('\n')
files_det = ['Length Date Time Name']
return files_det + dat
except Exception as exp:
msg = 'Unzipping Error with OS unzip utility'
logger.exception(msg)
append_scan_status(checksum, msg, repr(exp))
return []
def lipo_thin(checksum, src, dst):
@@ -347,7 +433,7 @@ def strings_and_entropies(checksum, src, exts):
'secrets': set(),
}
try:
if not src.exists():
if not (src and src.exists()):
return data
excludes = ('\\u0', 'com.google.')
eslash = ('Ljava', 'Lkotlin', 'kotlin', 'android')
+3
View File
@@ -55,6 +55,7 @@ from mobsf.StaticAnalyzer.views.common.appsec import (
)
from mobsf.StaticAnalyzer.views.common.async_task import (
async_analysis,
enqueued_task_init,
update_enqueued_task,
)
from mobsf.MalwareAnalyzer.views.MalwareDomainCheck import (
@@ -169,6 +170,7 @@ def ipa_analysis_task(checksum, app_dic, rescan, queue=False):
try:
if queue:
settings.ASYNC_ANALYSIS = True
enqueued_task_init(checksum)
append_scan_status(checksum, 'init')
msg = 'iOS Binary (IPA) Analysis Started'
logger.info(msg)
@@ -273,6 +275,7 @@ def ios_analysis_task(checksum, app_dic, rescan, queue=False):
try:
if queue:
settings.ASYNC_ANALYSIS = True
enqueued_task_init(checksum)
logger.info('iOS Source Code Analysis Started')
get_size_and_hashes(app_dic)
@@ -4,6 +4,7 @@
import logging
import os
from plistlib import (
FMT_XML,
dumps,
load,
loads,
@@ -13,12 +14,6 @@ from re import sub
from openstep_parser import OpenStepDecoder
from biplist import (
InvalidPlistException,
readPlist,
writePlistToString,
)
from mobsf.MobSF.utils import (
append_scan_status,
find_key_in_dict,
@@ -48,7 +43,7 @@ def get_bundle_id(pobj, src):
Look up in Info.plist, entitlements, pbxproj
"""
possible_ids = set()
skip_chars = {'$(', '${'}
skip_chars = ('$(', '${')
# From old Info.plist
bundle_id_og = pobj.get('CFBundleIdentifier', '')
@@ -101,10 +96,13 @@ def get_bundle_id(pobj, src):
def convert_bin_xml(bin_xml_file):
"""Convert Binary XML to Readable XML."""
try:
plist_obj = readPlist(bin_xml_file)
data = writePlistToString(plist_obj)
return data
except InvalidPlistException:
with open(bin_xml_file, 'rb') as fp:
plist_obj = load(fp)
# Serializing the plist object to a binary plist string
data = dumps(plist_obj, fmt=FMT_XML)
Path(bin_xml_file).write_bytes(data)
except Exception:
logger.warning('Failed to convert plist')
@@ -5,10 +5,9 @@ import json
import logging
import ntpath
import os
import plistlib
from pathlib import Path
import biplist
from django.conf import settings
from django.http import HttpResponseRedirect
from django.shortcuts import render
@@ -108,13 +107,17 @@ def run(request, api=False):
elif typ == 'plist':
file_format = 'json'
try:
dat = biplist.readPlist(sfile)
dat = json.dumps(dat, indent=4, sort_keys=True)
except biplist.InvalidPlistException:
with open(sfile, 'rb') as f:
# Attempt to load the plist, binary or XML
dat = plistlib.load(f)
# Convert the plist data to JSON for output
dat = json.dumps(dat, indent=4, sort_keys=True)
except plistlib.InvalidFileException:
# Handle invalid plist files (e.g., if it isn't binary or XML)
file_format = 'xml'
dat = Path(sfile).read_text()
except Exception:
pass
dat = None
elif typ == 'db':
file_format = 'asciidoc'
sql_dump = read_sqlite(sfile)
+10
View File
@@ -26,6 +26,16 @@
body {
zoom: 0.8;
}
.modal-backdrop {
position: fixed;
top: 0;
left: 0;
width: 100%;
height: 100%;
z-index: 1040; /* Ensure it is above most elements but below modal */
background-color: rgba(0, 0, 0, 0.5); /* Default backdrop color */
}
</style>
{% block extra_css %}
+68 -40
View File
@@ -50,8 +50,7 @@
<tr>
<th>Scan Task</th>
<th>Filename</th>
<th>Queued At</th>
<th>Completed At</th>
<th>Timeline</th>
<th>Status</th>
</tr>
</thead>
@@ -123,25 +122,6 @@
statusCell.appendChild(reportLink);
}
function updateCellStatus(task, statusCell, completedCell) {
statusCell.textContent = task.status.status || task.status;
if (!task.completed_at) {
addLoader(statusCell);
} else {
completedCell.textContent = new Date(task.completed_at).toLocaleString();
if (task.status === "Success") {
statusCell.classList.add("text-success");
statusCell.classList.remove("text-warning");
addReportLink(task, statusCell);
} else if (task.app_name === "Failed") {
statusCell.classList.add("text-danger");
statusCell.classList.remove("text-warning");
}
}
addAppName(task, statusCell);
}
function buildTableRow(task) {
const row = document.createElement("tr");
row.setAttribute("data-task-id", task.task_id);
@@ -151,10 +131,14 @@
const taskP = document.createElement("p");
const checksumP = document.createElement("p");
taskP.innerHTML = '<strong>Task ID: </strong> ';
const taskLabel = document.createElement("strong");
taskLabel.textContent = "Task ID: ";
taskP.appendChild(taskLabel);
taskP.appendChild(document.createTextNode(task.task_id));
checksumP.innerHTML = '<strong>Checksum: </strong> ';
const checksumLabel = document.createElement("strong");
checksumLabel.textContent = "Checksum: ";
checksumP.appendChild(checksumLabel);
checksumP.appendChild(document.createTextNode(task.checksum));
taskIdCell.appendChild(taskP);
@@ -164,33 +148,76 @@
const fileNameCell = document.createElement("td");
fileNameCell.textContent = task.file_name;
// Queued At
const queuedAtCell = document.createElement("td");
const date = new Date(task.created_at);
queuedAtCell.textContent = date.toLocaleString();
// Completed At
const completedCell = document.createElement("td");
completedCell.setAttribute("id", `completed-${task.task_id}`);
completedCell.textContent = task.completed_at ? new Date(task.completed_at).toLocaleString() : null;
// Timeline
const timelineCell = document.createElement("td");
timelineCell.setAttribute("id", `timeline-${task.task_id}`);
// Status
const statusCell = document.createElement("td");
statusCell.setAttribute("id", `status-${task.task_id}`);
statusCell.classList.add(task.completed_at ? "text-success" : "text-warning");
updateCellStatus(task, statusCell, completedCell);
updateCellStatus(task, statusCell, timelineCell);
// Append cells to row
row.appendChild(taskIdCell);
row.appendChild(fileNameCell);
row.appendChild(queuedAtCell);
row.appendChild(completedCell);
row.appendChild(timelineCell);
row.appendChild(statusCell);
return row;
}
function updateCellStatus(task, statusCell, timelineCell) {
// Update the status text
statusCell.textContent = task.status;
// Clear existing content in timelineCell
while (timelineCell.firstChild) {
timelineCell.removeChild(timelineCell.firstChild);
}
// Create and append "Queued At"
const queuedAtElement = document.createElement("p");
const queuedAtLabel = document.createElement("strong");
queuedAtLabel.textContent = "Queued At: ";
queuedAtElement.appendChild(queuedAtLabel);
queuedAtElement.appendChild(document.createTextNode(task.created_at ? new Date(task.created_at).toLocaleString() : "N/A"));
timelineCell.appendChild(queuedAtElement);
// Create and append "Started At"
const startedAtElement = document.createElement("p");
const startedAtLabel = document.createElement("strong");
startedAtLabel.textContent = "Started At: ";
startedAtElement.appendChild(startedAtLabel);
startedAtElement.appendChild(document.createTextNode(task.started_at ? new Date(task.started_at).toLocaleString() : "N/A"));
timelineCell.appendChild(startedAtElement);
// Create and append "Completed At"
const completedAtElement = document.createElement("p");
const completedAtLabel = document.createElement("strong");
completedAtLabel.textContent = "Completed At: ";
completedAtElement.appendChild(completedAtLabel);
completedAtElement.appendChild(document.createTextNode(task.completed_at ? new Date(task.completed_at).toLocaleString() : "N/A"));
timelineCell.appendChild(completedAtElement);
// Add loader or success indicators
if (!task.completed_at) {
addLoader(statusCell);
} else {
if (task.status === "Success") {
statusCell.classList.add("text-success");
statusCell.classList.remove("text-warning");
addReportLink(task, statusCell);
} else if (task.status === "Failed") {
statusCell.classList.add("text-danger");
statusCell.classList.remove("text-warning");
}
}
addAppName(task, statusCell);
}
async function renderTable() {
const tasks = await fetchTasks();
tasksTableBody.innerHTML = ""; // Clear existing rows
@@ -198,7 +225,7 @@
if (tasks.length === 0) {
const noTasksRow = document.createElement("tr");
const noTasksCell = document.createElement("td");
noTasksCell.setAttribute("colspan", "5");
noTasksCell.setAttribute("colspan", "4");
noTasksCell.textContent = "No tasks in queue.";
noTasksRow.appendChild(noTasksCell);
tasksTableBody.appendChild(noTasksRow);
@@ -217,11 +244,11 @@
tasks.forEach((task) => {
const statusCell = document.getElementById(`status-${task.task_id}`);
const completedCell = document.getElementById(`completed-${task.task_id}`);
const timelineCell = document.getElementById(`timeline-${task.task_id}`);
if (statusCell && completedCell) {
// Update existing task status
updateCellStatus(task, statusCell, completedCell);
if (statusCell && timelineCell) {
// Update existing task status and timeline
updateCellStatus(task, statusCell, timelineCell);
} else if (!existingTaskIds.includes(task.task_id)) {
// Add a new row if the task ID is not found, at the top of the table
const newRow = buildTableRow(task);
@@ -231,6 +258,7 @@
}
// Initial render
renderTable();
@@ -2354,9 +2354,11 @@
<strong><i class="fa fa-archive"></i> SBOM</strong>
</p>
<div class="list-group">
{% if sbom %}
{% include 'base/list.html' with list=sbom.sbom_versioned type="Versioned Packages" limit=100 %}
{% include 'base/list.html' with list=sbom.sbom_packages type="Packages" limit=100 %}
</div>
{% endif %}
</div>
</div>
</div>
</div><!-- /.card -->
@@ -2555,11 +2557,13 @@ imageSeries.data = [
{% endblock %}
{% block scan_logs %}
<table id="scan_logs" class="table table-bordered table-hover table-striped">
<thead>
<tr>
<th>Timestamp</th>
<th>Event</th>
<th>Error</th>
</tr>
</thead>
<tbody>
{% for log in logs %}
<tr>
@@ -1762,9 +1762,11 @@
<strong><i class="fa fa-archive"></i> SBOM</strong>
</p>
<div class="list-group">
{% if sbom %}
{% include 'base/list.html' with list=sbom.sbom_versioned type="Versioned Packages" limit=100 %}
{% include 'base/list.html' with list=sbom.sbom_packages type="Packages" limit=100 %}
</div>
{% endif %}
</div>
</div>
</div>
</div><!-- /.card -->
@@ -1942,11 +1944,13 @@
{% endblock %}
{% block scan_logs %}
<table id="scan_logs" class="table table-bordered table-hover table-striped">
<thead>
<tr>
<th>Timestamp</th>
<th>Event</th>
<th>Error</th>
</tr>
</thead>
<tbody>
{% for log in logs %}
<tr>
@@ -1921,12 +1921,14 @@
{% endblock %}
{% block scan_logs %}
<table id="scan_logs" class="table table-bordered table-hover table-striped">
<tr>
<th>Timestamp</th>
<th>Event</th>
<th>Error</th>
</tr>
<tbody>
<thead>
<tr>
<th>Timestamp</th>
<th>Event</th>
<th>Error</th>
</tr>
</thead>
<tbody>
{% for log in logs %}
<tr>
<td>
@@ -1392,11 +1392,13 @@
{% endblock %}
{% block scan_logs %}
<table id="scan_logs" class="table table-bordered table-hover table-striped">
<thead>
<tr>
<th>Timestamp</th>
<th>Event</th>
<th>Error</th>
</tr>
</thead>
<tbody>
{% for log in logs %}
<tr>
@@ -463,11 +463,13 @@
{% endblock %}
{% block scan_logs %}
<table id="scan_logs" class="table table-bordered table-hover table-striped">
<thead>
<tr>
<th>Timestamp</th>
<th>Event</th>
<th>Error</th>
</tr>
</thead>
<tbody>
{% for log in logs %}
<tr>
Generated
+12 -11
View File
@@ -229,16 +229,6 @@ files = [
{file = "billiard-4.2.1.tar.gz", hash = "sha256:12b641b0c539073fc8d3f5b8b7be998956665c4233c7c1fcd66a7e677c4fb36f"},
]
[[package]]
name = "biplist"
version = "1.0.3"
description = "biplist is a library for reading/writing binary plists."
optional = false
python-versions = "*"
files = [
{file = "biplist-1.0.3.tar.gz", hash = "sha256:4c0549764c5fe50b28042ec21aa2e14fe1a2224e239a1dae77d9e7f3932aa4c6"},
]
[[package]]
name = "blinker"
version = "1.9.0"
@@ -651,6 +641,17 @@ ssh = ["bcrypt (>=3.1.5)"]
test = ["certifi", "cryptography-vectors (==43.0.3)", "pretend", "pytest (>=6.2.0)", "pytest-benchmark", "pytest-cov", "pytest-xdist"]
test-randomorder = ["pytest-randomly"]
[[package]]
name = "defusedxml"
version = "0.7.1"
description = "XML bomb protection for Python stdlib modules"
optional = false
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
files = [
{file = "defusedxml-0.7.1-py2.py3-none-any.whl", hash = "sha256:a352e7e428770286cc899e2542b6cdaedb2b4953ff269a210103ec58f6198a61"},
{file = "defusedxml-0.7.1.tar.gz", hash = "sha256:1bb3032db185915b62d7c6209c5a8792be6a32ab2fedacc84e01b52c51aa3e69"},
]
[[package]]
name = "distro"
version = "1.9.0"
@@ -2666,4 +2667,4 @@ cffi = ["cffi (>=1.11)"]
[metadata]
lock-version = "2.0"
python-versions = "^3.10"
content-hash = "47edc939d8d6d36767bbc3210b4440106dbda32948de859d18a100e458815e0c"
content-hash = "b89f2cedf9ba39c6ec2d9ef024826fb8af034db78b2894597328aa292b419fdb"
+2 -2
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "mobsf"
version = "4.2.7"
version = "4.2.8"
description = "Mobile Security Framework (MobSF) is an automated, all-in-one mobile application (Android/iOS/Windows) pen-testing, malware analysis and security assessment framework capable of performing static and dynamic analysis."
keywords = ["mobsf", "mobile security framework", "mobile security", "security tool", "static analysis", "dynamic analysis", "malware analysis"]
authors = ["Ajin Abraham <[email protected]>"]
@@ -27,7 +27,6 @@ python = "^3.10"
django = ">=3.1.5"
lxml = ">=4.6.2"
rsa = ">=4.7"
biplist = ">=1.0.3"
requests = ">=2.25.1"
bs4 = ">=0.0.1"
colorlog = ">=4.7.2"
@@ -61,6 +60,7 @@ lief = "^0.15.1"
packaging = ">=21.3"
django-ratelimit = "^4.1.0"
django-q2 = "^1.7.4"
defusedxml = "^0.7.1"
[build-system]
requires = ["poetry-core"]
+4
View File
@@ -1,4 +1,8 @@
#!/bin/bash
JDK_FILE=openjdk-22.0.2_linux-x64_bin.tar.gz
JDK_FILE_ARM=openjdk-22.0.2_linux-aarch64_bin.tar.gz
WKH_FILE=wkhtmltox_0.12.6.1-3.bookworm_amd64.deb
WKH_FILE_ARM=wkhtmltox_0.12.6.1-3.bookworm_arm64.deb
# For apktool
mkdir -p /home/mobsf/.local/share/apktool/framework