Multiple QA

mobsf/MalwareAnalyzer/views/android/apkid.py

@@ -6,6 +6,7 @@
 
 from mobsf.MobSF.utils import (
     append_scan_status,
+    run_with_timeout,
     settings_enabled,
 )
 
@@ -49,13 +50,28 @@ def apkid_analysis(checksum, apk_file):
     )
     rules = options.rules_manager.load()
     scanner = Scanner(rules, options)
-    res = scanner.scan_file(apk_file)
+    findings = {}
+    res = None
     try:
-        findings = output._build_json_output(res)['files']
+        res = run_with_timeout(
+            scanner.scan_file,
+            settings.BINARY_ANALYSIS_TIMEOUT,
+            apk_file)
+    except Exception as e:
+        msg = 'APKID scan timed out'
+        logger.error(msg)
+        append_scan_status(
+            checksum,
+            msg,
+            str(e))
+    try:
+        if res:
+            findings = output._build_json_output(res)['files']
     except AttributeError:
         # apkid >= 2.0.3
         try:
-            findings = output.build_json_output(res)['files']
+            if res:
+                findings = output.build_json_output(res)['files']
         except AttributeError:
             msg = (
                 'yara-python dependency required by '
@@ -66,7 +82,6 @@ def apkid_analysis(checksum, apk_file):
                 checksum,
                 msg,
                 'Missing dependency')
-            findings = {}
     sanitized = {}
     for item in findings:
         filename = item['filename']

mobsf/MobSF/init.py

@@ -16,7 +16,7 @@
 
 logger = logging.getLogger(__name__)
 
-VERSION = '4.1.0'
+VERSION = '4.1.1'
 BANNER = r"""
   __  __       _    ____  _____       _  _    _ 
  |  \/  | ___ | |__/ ___||  ___|_   _| || |  / |

mobsf/MobSF/settings.py

@@ -330,7 +330,9 @@
         },
     },
 }
-JADX_TIMEOUT = int(os.getenv('MOBSF_JADX_TIMEOUT', 1800))
+JADX_TIMEOUT = int(os.getenv('MOBSF_JADX_TIMEOUT', 1000))
+SAST_TIMEOUT = int(os.getenv('MOBSF_SAST_TIMEOUT', 1000))
+BINARY_ANALYSIS_TIMEOUT = int(os.getenv('MOBSF_BINARY_ANALYSIS_TIMEOUT', 600))
 DISABLE_AUTHENTICATION = os.getenv('MOBSF_DISABLE_AUTHENTICATION')
 RATELIMIT = os.getenv('MOBSF_RATELIMIT', '7/m')
 USE_X_FORWARDED_HOST = bool(

mobsf/MobSF/urls.py

@@ -85,6 +85,7 @@
     # Static Analysis
     re_path(r'^api/v1/upload$', api_sz.api_upload),
     re_path(r'^api/v1/scan$', api_sz.api_scan),
+    re_path(r'^api/v1/search$', api_sz.api_search),
     re_path(r'^api/v1/scan_logs$', api_sz.api_scan_logs),
     re_path(r'^api/v1/delete_scan$', api_sz.api_delete_scan),
     re_path(r'^api/v1/download_pdf$', api_sz.api_pdf_report),
@@ -198,7 +199,6 @@
         re_path(r'^search$', home.search),
         re_path(r'^status/$', home.scan_status, name='status'),
         re_path(r'^error/$', home.error, name='error'),
-        re_path(r'^not_found/$', home.not_found),
         re_path(r'^zip_format/$', home.zip_format),
         re_path(r'^dynamic_analysis/$', home.dynamic_analysis, name='dynamic'),
 

mobsf/MobSF/utils.py

@@ -947,3 +947,28 @@ def get_scan_logs(checksum):
         msg = 'Fetching scan logs from the DB failed.'
         logger.exception(msg)
     return []
+
+
+class TaskTimeoutError(Exception):
+    pass
+
+
+def run_with_timeout(func, limit, *args, **kwargs):
+    def run_func(result, *args, **kwargs):
+        result.append(func(*args, **kwargs))
+
+    result = []
+    thread = threading.Thread(
+        target=run_func,
+        args=(result, *args),
+        kwargs=kwargs)
+    thread.start()
+    thread.join(limit)
+
+    if thread.is_alive():
+        msg = (f'function <{func.__name__}> '
+               f'timed out after {limit} seconds')
+        raise TaskTimeoutError(msg)
+    if result and len(result) > 0:
+        return result[0]
+    return None

mobsf/MobSF/views/api/api_static_analysis.py

@@ -12,7 +12,12 @@
     is_md5,
 )
 from mobsf.MobSF.views.helpers import request_method
-from mobsf.MobSF.views.home import RecentScans, Upload, delete_scan
+from mobsf.MobSF.views.home import (
+    RecentScans,
+    Upload,
+    delete_scan,
+    search,
+)
 from mobsf.MobSF.views.api.api_middleware import make_api_response
 from mobsf.StaticAnalyzer.views.android.views import view_source
 from mobsf.StaticAnalyzer.views.android.static_analyzer import static_analyzer
@@ -180,6 +185,21 @@ def api_json_report(request):
     return response
 
 
+@request_method(['POST'])
+@csrf_exempt
+def api_search(request):
+    """Search by checksum or text."""
+    if 'query' not in request.POST:
+        return make_api_response(
+            {'error': 'Missing Parameters'}, 422)
+    resp = search(request, api=True)
+    if 'checksum' in resp:
+        request.POST = {'hash': resp['checksum']}
+        return api_json_report(request)
+    elif 'error' in resp:
+        return make_api_response(resp, 404)
+
+
 @request_method(['POST'])
 @csrf_exempt
 def api_view_source(request):

mobsf/MobSF/views/home.py

@@ -22,6 +22,7 @@
 
 from mobsf.MobSF.forms import FormUtil, UploadFileForm
 from mobsf.MobSF.utils import (
+    MD5_REGEX,
     api_key,
     get_md5,
     is_dir_exists,
@@ -244,16 +245,6 @@
     return render(request, template, context)
 
 
-def not_found(request, *args):
-    """Not Found Route."""
-    context = {
-        'title': 'Not Found',
-        'version': settings.MOBSF_VER,
-    }
-    template = 'general/not_found.html'
-    return render(request, template, context)
-
-
 @login_required
 def dynamic_analysis(request):
     """Dynamic Analysis Landing."""
@@ -337,18 +328,43 @@
 
 
 @login_required
-def search(request):
-    """Search Scan by MD5 Route."""
-    md5 = request.GET['md5']
-    if re.match('[0-9a-f]{32}', md5):
-        db_obj = RecentScansDB.objects.filter(MD5=md5)
-        if db_obj.exists():
-            e = db_obj[0]
-            url = f'/{e.ANALYZER}/{e.MD5}/'
-            return HttpResponseRedirect(url)
-        else:
-            return HttpResponseRedirect('/not_found/')
-    return print_n_send_error_response(request, 'Invalid Scan Hash')
+def search(request, api=False):
+    """Search scan by checksum or text."""
+    if request.method == 'POST':
+        query = request.POST['query']
+    else:
+        query = request.GET['query']
+
+    if not query:
+        msg = 'No search query provided.'
+        return print_n_send_error_response(request, msg, api)
+
+    checksum = query if re.match(MD5_REGEX, query) else find_checksum(query)
+
+    if checksum and re.match(MD5_REGEX, checksum):
+        db_obj = RecentScansDB.objects.filter(MD5=checksum).first()
+        if db_obj:
+            url = f'/{db_obj.ANALYZER}/{db_obj.MD5}/'
+            if api:
+                return {'checksum': db_obj.MD5}
+            else:
+                return HttpResponseRedirect(url)
+
+    msg = 'You can search by MD5, app name, package name, or file name.'
+    return print_n_send_error_response(request, msg, api, 'Scan not found')
+
+
+def find_checksum(query):
+    """Get the first matching checksum from the database."""
+    search_fields = ['FILE_NAME', 'PACKAGE_NAME', 'APP_NAME']
+
+    for field in search_fields:
+        result = RecentScansDB.objects.filter(
+            **{f'{field}__icontains': query}).first()
+        if result:
+            return result.MD5
+
+    return None
 
 # AJAX
 
@@ -453,7 +469,7 @@
             else:
                 md5_hash = request.POST['md5']
             data = {'deleted': 'scan hash not found'}
-            if re.match('[0-9a-f]{32}', md5_hash):
+            if re.match(MD5_REGEX, md5_hash):
                 # Delete DB Entries
                 scan = RecentScansDB.objects.filter(MD5=md5_hash)
                 if scan.exists():
@@ -485,10 +501,7 @@
     except Exception as exp:
         msg = str(exp)
         exp_doc = exp.__doc__
-        if api:
-            return print_n_send_error_response(request, msg, True, exp_doc)
-        else:
-            return print_n_send_error_response(request, msg, False, exp_doc)
+        return print_n_send_error_response(request, msg, api, exp_doc)
 
 
 class RecentScans(object):

mobsf/StaticAnalyzer/tests.py

@@ -117,7 +117,7 @@ def static_analysis_test():
             logger.info(resp.content)
             return True
 
-        # Search by MD5
+        # Search by MD5 and text
         if platform.system() in ['Darwin', 'Linux']:
             scan_md5s = ['02e7989c457ab67eb514a8328779f256',
                          '82ab8b2193b3cfb1c737e3a786be363a',
@@ -133,18 +133,23 @@ def static_analysis_test():
                          '57bb5be0ea44a755ada4a93885c3825e',
                          '8179b557433835827a70510584f3143e',
                          '7b0a23bffc80bac05739ea1af898daad']
+        # Search by text
+        queries = [
+            'diva',
+            'webview',
+        ]
         logger.info('Running Search test')
-        for scan_md5 in scan_md5s:
-            url = '/search?md5={}'.format(scan_md5)
+        for q in scan_md5s + queries:
+            url = f'/search?query={q}'
             resp = http_client.get(url, follow=True)
             assert (resp.status_code == 200)
             if resp.status_code == 200:
-                logger.info('[OK] Search by MD5 test passed for %s', scan_md5)
+                logger.info('[OK] Search by query test passed for %s', q)
             else:
-                logger.error('Search by MD5 test failed for %s', scan_md5)
+                logger.error('Search by query test failed for %s', q)
                 logger.info(resp.content)
                 return True
-        logger.info('[OK] Search by MD5 tests completed')
+        logger.info('[OK] Search by MD5 and text tests completed')
 
         # Deleting Scan Results
         logger.info('Running Delete Scan Results test')
@@ -256,6 +261,18 @@ def api_test():
                 logger.error('Scan Logs API test: %s', upl['hash'])
                 return True
         logger.info('[OK] Static Analysis API test completed')
+        # Search API Tests
+        logger.info('Running Search API tests')
+        for term in ['diva', 'webview', '52c50ae824e329ba8b5b7a0f523efffe']:
+            resp = http_client.post(
+                '/api/v1/search',
+                {'query': term},
+                HTTP_AUTHORIZATION=auth)
+            if resp.status_code == 200:
+                logger.info('[OK] Search API test: %s', term)
+            else:
+                logger.error('Search API test: %s', term)
+                return True
         # PDF Tests
         logger.info('Running PDF Generation API Test')
         if platform.system() in ['Darwin', 'Linux']:

mobsf/StaticAnalyzer/views/common/binary/elf.py

@@ -8,6 +8,11 @@
 from mobsf.StaticAnalyzer.views.common.binary.strings import (
     strings_on_binary,
 )
+from mobsf.MobSF.utils import (
+    run_with_timeout,
+)
+
+from django.conf import settings
 
 
 NA = 'Not Applicable'
@@ -32,7 +37,10 @@ class ELFChecksec:
     def __init__(self, elf_file, so_rel):
         self.elf_path = elf_file.as_posix()
         self.elf_rel = so_rel
-        self.elf = lief.parse(self.elf_path)
+        self.elf = run_with_timeout(
+            lief.parse,
+            settings.BINARY_ANALYSIS_TIMEOUT,
+            self.elf_path)
 
     def checksec(self):
         elf_dict = {}

mobsf/StaticAnalyzer/views/common/binary/lib_analysis.py

@@ -27,6 +27,9 @@ def library_analysis(checksum, src, arch):
         f'{arch}_analysis': [],
         f'{arch}_strings': [],
         f'{arch}_symbols': [],
+        'framework_analysis': [],
+        'framework_strings': [],
+        'framework_symbols': [],
     }
     try:
         if arch == 'macho':
@@ -79,9 +82,6 @@ def library_analysis(checksum, src, arch):
                     rel_path: symbols})
         if ext == '*.dylib':
             # Do Framework Analysis for iOS
-            res['framework_analysis'] = []
-            res['framework_strings'] = []
-            res['framework_symbols'] = []
             frameworks_analysis(checksum, src, base_dir, res)
             if res['framework_strings']:
                 res[f'{arch}_strings'].extend(

mobsf/StaticAnalyzer/views/common/binary/macho.py

@@ -9,6 +9,11 @@
 from mobsf.StaticAnalyzer.views.common.binary.strings import (
     strings_on_binary,
 )
+from mobsf.MobSF.utils import (
+    run_with_timeout,
+)
+
+from django.conf import settings
 
 
 def objdump_is_debug_symbol_stripped(macho_file):
@@ -28,7 +33,10 @@ def __init__(self, macho, rel_path=None):
             self.macho_name = rel_path
         else:
             self.macho_name = macho.name
-        self.macho = lief.parse(self.macho_path)
+        self.macho = run_with_timeout(
+            lief.parse,
+            settings.BINARY_ANALYSIS_TIMEOUT,
+            self.macho_path)
 
     def checksec(self):
         macho_dict = {}