TLS/SSL Security Tester (#1726)

* TLS/SSL Security Tester
* Frida Hooking Code QA
* Fix typos in permission mapping
* Xposed improvements
* Dynamic Analyzer QA
* Logcat app only logs refactoring
* Updated and exposed new REST APIs
* LGTM Findings Review and override
* Python and JavaScript code QA and lib update

mobsf/DynamicAnalyzer/tools/frida_scripts/default/ssl_pinning_bypass.js

@@ -139,10 +139,11 @@ Java.perform(function() {
     /*** HttpsURLConnection ***/
     try {
         var HttpsURLConnection = Java.use('javax.net.ssl.HttpsURLConnection');
+        /*
         HttpsURLConnection.setDefaultHostnameVerifier.implementation = function(hostnameVerifier) {
             send('[SSL Pinning Bypass] HttpsURLConnection.setDefaultHostnameVerifier bypassed');
             return null;
-        };
+        };*/
         HttpsURLConnection.setSSLSocketFactory.implementation = function(SSLSocketFactory) {
             send('[SSL Pinning Bypass] HttpsURLConnection.setSSLSocketFactory bypassed');
             return null;

mobsf/DynamicAnalyzer/tools/frida_scripts/others/bypass_method.js

@@ -0,0 +1,8 @@
+Java.perform(function () {
+    send('Installing Bypass');
+    var hook = Java.use('some.package.Activity');
+    hook.someMethod.overload().implementation = function () {
+        send('--------------Bypassed!-------');
+        return false;
+    };
+});

mobsf/DynamicAnalyzer/tools/webproxy.py

@@ -6,6 +6,8 @@
 
 import requests
 
+from django.conf import settings
+
 from mobsf.MobSF.utils import is_file_exists, upstream_proxy
 
 logger = logging.getLogger(__name__)
@@ -70,3 +72,18 @@ def get_ca_file():
     if not is_file_exists(ca_file):
         create_ca()
     return ca_file
+
+
+def get_traffic(package):
+    web = Path.home() / '.httptools' / 'flows' / f'{package}.flow.txt'
+    if web.is_file():
+        return web.read_text('utf-8', 'ignore')
+    return ''
+
+
+def get_http_tools_url(req):
+    """Get httptools URL from request."""
+    scheme = req.scheme
+    ip = req.get_host().split(':')[0]
+    port = settings.PROXY_PORT
+    return f'{scheme}://{ip}:{str(port)}'

mobsf/DynamicAnalyzer/views/android/analysis.py

@@ -5,6 +5,7 @@
 import re
 import shutil
 import tarfile
+from json import load
 from pathlib import Path
 
 from mobsf.MobSF.utils import (
@@ -67,6 +68,7 @@ def run_analysis(apk_dir, md5_hash, package):
     analysis_result['xml'] = all_files['xml']
     analysis_result['sqlite'] = all_files['sqlite']
     analysis_result['other_files'] = all_files['others']
+    analysis_result['tls_tests'] = get_tls_logs(apk_dir, md5_hash)
     return analysis_result
 
 
@@ -112,6 +114,15 @@ def get_screenshots(md5_hash, download_dir):
     return result
 
 
+def get_tls_logs(apk_dir, md5_hash):
+    """Get TLS/SSL test logs."""
+    out = Path(apk_dir) / 'mobsf_tls_tests.json'
+    if not out.exists():
+        return None
+    with out.open(encoding='utf-8') as src:
+        return load(src)
+
+
 def get_log_data(apk_dir, package):
     """Get Data for analysis."""
     logcat_data = []
@@ -133,26 +144,26 @@ def get_log_data(apk_dir, package):
                      errors='ignore') as flip:
             web_data = flip.read()
     if is_file_exists(logcat):
-        with io.open(logcat,
+        with io.open(logcat,  # lgtm [py/path-injection]
                      mode='r',
                      encoding='utf8',
                      errors='ignore') as flip:
             logcat_data = flip.readlines()
             traffic = ''.join(logcat_data)
     if is_file_exists(xlogcat):
-        with io.open(xlogcat,
+        with io.open(xlogcat,  # lgtm [py/path-injection]
                      mode='r',
                      encoding='utf8',
                      errors='ignore') as flip:
             droidmon_data = flip.read()
     if is_file_exists(apimon):
-        with io.open(apimon,
+        with io.open(apimon,  # lgtm [py/path-injection]
                      mode='r',
                      encoding='utf8',
                      errors='ignore') as flip:
             apimon_data = flip.read()
     if is_file_exists(fd_logs):
-        with io.open(fd_logs,
+        with io.open(fd_logs,  # lgtm [py/path-injection]
                      mode='r',
                      encoding='utf8',
                      errors='ignore') as flip:
@@ -206,7 +217,7 @@ def get_app_files(apk_dir, md5_hash, package):
                         all_files['xml'].append(
                             {'type': 'xml', 'file': fileparam})
                     else:
-                        with open(file_path,
+                        with open(file_path,  # lgtm [py/path-injection]
                                   'r',
                                   encoding='ISO-8859-1') as flip:
                             file_cnt_sig = flip.read(6)

mobsf/DynamicAnalyzer/views/android/dynamic_analyzer.py

@@ -20,13 +20,13 @@
     strict_package_check,
 )
 from mobsf.DynamicAnalyzer.tools.webproxy import (
+    get_http_tools_url,
     start_httptools_ui,
     stop_httptools,
 )
 from mobsf.MobSF.utils import (
     get_config_loc,
     get_device,
-    get_http_tools_url,
     get_proxy_ip,
     is_md5,
     print_n_send_error_response,
@@ -42,6 +42,8 @@ def dynamic_analysis(request, api=False):
     try:
         scan_apps = []
         device_packages = {}
+        and_ver = None
+        and_sdk = None
         apks = StaticAnalyzerAndroid.objects.filter(
             APP_TYPE='apk')
         for apk in reversed(apks):
@@ -64,19 +66,22 @@ def dynamic_analysis(request, api=False):
                    ' set ANALYZER_IDENTIFIER in '
                    f'{get_config_loc()}')
             return print_n_send_error_response(request, msg, api)
-        proxy_ip = get_proxy_ip(identifier)
         try:
             if identifier:
                 env = Environment(identifier)
                 device_packages = env.get_device_packages()
                 pkg_file = Path(settings.DWD_DIR) / 'packages.json'
                 with pkg_file.open('w', encoding='utf-8') as target:
                     dump(device_packages, target)
+                and_ver = env.get_android_version()
+                and_sdk = env.get_android_sdk()
         except Exception:
             pass
         context = {'apps': scan_apps,
                    'identifier': identifier,
-                   'proxy_ip': proxy_ip,
+                   'android_version': and_ver,
+                   'android_sdk': and_sdk,
+                   'proxy_ip': get_proxy_ip(identifier),
                    'proxy_port': settings.PROXY_PORT,
                    'settings_loc': get_config_loc(),
                    'device_packages': device_packages,
@@ -95,7 +100,6 @@ def dynamic_analysis(request, api=False):
 
 def dynamic_analyzer(request, checksum, api=False):
     """Android Dynamic Analyzer Environment."""
-    logger.info('Creating Dynamic Analysis Environment')
     try:
         identifier = None
         if api:
@@ -117,6 +121,7 @@ def dynamic_analyzer(request, checksum, api=False):
                 request,
                 'Cannot get package name from checksum',
                 api)
+        logger.info('Creating Dynamic Analysis Environment for %s', package)
         try:
             identifier = get_device()
         except Exception:
@@ -137,7 +142,7 @@ def dynamic_analyzer(request, checksum, api=False):
         logger.info('Android Version identified as %s', version)
         xposed_first_run = False
         if not env.is_mobsfyied(version):
-            msg = ('This Android instance is not MobSfyed/Outdated.\n'
+            msg = ('This Android instance is not MobSFyed/Outdated.\n'
                    'MobSFying the android runtime environment')
             logger.warning(msg)
             if not env.mobsfy_init():
@@ -217,7 +222,8 @@ def httptools_start(request):
         else:
             project = ''
         url = f'{httptools_url}/dashboard/{project}'
-        return HttpResponseRedirect(url)  # lgtm [py/reflective-xss]
+        ret = HttpResponseRedirect(url)
+        return ret  # lgtm [py/reflective-xss] lgtm [py/url-redirection]
     except Exception:
         logger.exception('Starting httptools Web UI')
         err = 'Error Starting httptools UI'
@@ -254,7 +260,6 @@ def read_process():
                 while g.is_pending():
                     lines = g.readlines()
                     for _, line in lines:
-                        time.sleep(.01)
                         yield 'data:{}\n\n'.format(line)
             return StreamingHttpResponse(read_process(),
                                          content_type='text/event-stream')
@@ -302,8 +307,9 @@ def trigger_static_analysis(request, checksum):
         }
         add_to_recent_scan(data)
         return HttpResponseRedirect(
-            f'/static_analyzer/?name={package}.apk'
-            f'&checksum={checksum}&type=apk')
+            f'/static_analyzer/?name='
+            f'{package}.apk&checksum={checksum}'  # lgtm [py/url-redirection]
+            f'&type=apk')
     except Exception:
         msg = 'On device APK Static Analysis'
         logger.exception(msg)

mobsf/DynamicAnalyzer/views/android/environment.py

@@ -18,6 +18,7 @@
 
 from mobsf.DynamicAnalyzer.tools.webproxy import (
     get_ca_file,
+    get_http_tools_url,
     start_proxy,
     stop_httptools,
 )
@@ -27,7 +28,6 @@
 from mobsf.MobSF.utils import (
     get_adb,
     get_device,
-    get_http_tools_url,
     get_proxy_ip,
     is_file_exists,
     python_list,
@@ -61,10 +61,10 @@ def check_connect_error(self, output):
             return False
         return True
 
-    def run_subprocess_verify_output(self, command):
+    def run_subprocess_verify_output(self, cmd):
         """Run subprocess and verify execution."""
-        out = subprocess.check_output(command)
-        self.wait(2)
+        out = subprocess.check_output(cmd)  # lgtm [py/command-line-injection]
+        self.wait(2)                        # adb shell is allowed
         return self.check_connect_error(out)
 
     def connect_n_mount(self):
@@ -403,6 +403,13 @@ def get_android_arch(self):
             'ro.product.cpu.abi'], True)
         return out.decode('utf-8').rstrip()
 
+    def get_android_sdk(self):
+        """Get Android API version."""
+        out = self.adb_command([
+            'getprop',
+            'ro.build.version.sdk'], True)
+        return out.decode('utf-8').strip()
+
     def get_device_packages(self):
         """Get all packages from device."""
         device_packages = {}
@@ -425,6 +432,9 @@ def get_device_packages(self):
                 '-b',
                 apk], True)
             md5 = out1.decode('utf-8').strip()
+            if '.apk' in md5:
+                # -b not respected in Android 5.0
+                md5 = md5.split()[0]
             device_packages[md5] = (pkg, apk)
         return device_packages
 
@@ -458,14 +468,11 @@ def system_check(self, runtime):
         """Check if /system is writable."""
         try:
             try:
-                out = self.adb_command([
-                    'getprop',
-                    'ro.build.version.sdk'], True)
-                if out:
-                    api = int(out.decode('utf-8').strip())
+                api = self.get_android_sdk()
+                if api:
                     logger.info('Android API Level '
                                 'identified as %s', api)
-                    if api > ANDROID_API_SUPPORTED:
+                    if int(api) > ANDROID_API_SUPPORTED:
                         logger.error('This API Level is not supported'
                                      ' for Dynamic Analysis.')
                         return False
@@ -505,6 +512,15 @@ def launch_n_capture(self, package, activity, outfile):
         logger.info('Stopping app')
         self.adb_command(['am', 'force-stop', package], True)
 
+    def run_app(self, package):
+        """Launch an app with package name."""
+        self.adb_command(['monkey',
+                          '-p',
+                          package,
+                          '-c',
+                          'android.intent.category.LAUNCHER',
+                          '1'], True)
+
     def is_mobsfyied(self, android_version):
         """Check is Device is MobSFyed."""
         logger.info('Environment MobSFyed Check')
@@ -598,7 +614,7 @@ def xposed_setup(self, android_version):
                                   'JustTrustMe.apk')
         rootcloak = os.path.join(self.tools_dir,
                                  xposed_modules,
-                                 'RootCloak.apk')
+                                 'com.devadvance.rootcloak2_v18_c43b61.apk')
         proxyon = os.path.join(self.tools_dir,
                                xposed_modules,
                                'mobi.acpm.proxyon_v1_419b04.apk')

mobsf/DynamicAnalyzer/views/android/frida_core.py

@@ -45,6 +45,8 @@ def get_default_scripts(self):
         """Get default Frida Scripts."""
         combined_script = []
         header = []
+        if not self.defaults:
+            return header
         def_scripts = os.path.join(self.frida_dir, 'default')
         files = glob.glob(def_scripts + '**/*.js', recursive=True)
         for item in files:
@@ -58,6 +60,8 @@ def get_default_scripts(self):
     def get_auxiliary(self):
         """Get auxiliary hooks."""
         scripts = []
+        if not self.auxiliary:
+            return scripts
         for itm in self.auxiliary:
             if itm == 'enum_class':
                 scripts.append(get_loaded_classes())
@@ -75,9 +79,12 @@ def get_auxiliary(self):
 
     def get_script(self):
         """Get final script."""
-        scripts = self.get_default_scripts()
+        if not self.code:
+            self.code = ''
+        # Load custom code first
+        scripts = [self.code]
+        scripts.extend(self.get_default_scripts())
         scripts.extend(self.get_auxiliary())
-        scripts.extend([self.code])
         final = 'setTimeout(function() {{ {} }}, 0)'.format(
             '\n'.join(scripts))
         return final