[HOTFIX] + Features (#2444)

Add support for sample download in recent scans.
Bug fix in firebase analysis (dict mutation errors)

mobsf/MobSF/init.py

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

mobsf/MobSF/settings.py

@@ -73,13 +73,17 @@
     '.zip': 'application/zip',
     '.tar': 'application/x-tar',
     '.apk': 'application/octet-stream',
+    '.apks': 'application/octet-stream',
+    '.xapk': 'application/octet-stream',
+    '.aab': 'application/octet-stream',
     '.ipa': 'application/octet-stream',
     '.jar': 'application/java-archive',
     '.aar': 'application/octet-stream',
     '.so': 'application/octet-stream',
     '.dylib': 'application/octet-stream',
     '.a': 'application/octet-stream',
     '.pcap': 'application/vnd.tcpdump.pcap',
+    '.appx': 'application/vns.ms-appx',
 }
 # =============ALLOWED MIMETYPES=================
 APK_MIME = [

mobsf/MobSF/urls.py

@@ -184,6 +184,9 @@
         re_path(r'^$', home.index, name='home'),
         re_path(r'^upload/$', home.Upload.as_view, name='upload'),
         re_path(r'^download/', home.download, name='download'),
+        re_path(fr'^download_binary/{checksum_regex}/$',
+                home.download_binary,
+                name='download_binary'),
         re_path(r'^download_scan/', home.download_apk, name='download_scan'),
         re_path(r'^generate_downloads/$',
                 home.generate_download,

mobsf/MobSF/views/home.py

@@ -57,6 +57,8 @@
 
 LINUX_PLATFORM = ['Darwin', 'Linux']
 HTTP_BAD_REQUEST = 400
+HTTP_STATUS_404 = 404
+HTTP_SERVER_ERROR = 500
 logger = logging.getLogger(__name__)
 register.filter('key', key)
 
@@ -388,30 +390,82 @@ def scan_status(request, api=False):
     return send_response(data, api)
 
 
+def file_download(dwd_file, filename, content_type):
+    """HTTP file download response."""
+    with open(dwd_file, 'rb') as file:
+        wrapper = FileWrapper(file)
+        response = HttpResponse(wrapper, content_type=content_type)
+        response['Content-Length'] = dwd_file.stat().st_size
+        if filename:
+            val = f'attachment; filename="{filename}"'
+            response['Content-Disposition'] = val
+        return response
+
+
 @login_required
-def download(request):
-    """Download from mobsf.MobSF Route."""
-    if request.method == 'GET':
-        root = settings.DWD_DIR
+@require_http_methods(['GET'])
+def download_binary(request, checksum, api=False):
+    """Download binary from uploads directory."""
+    try:
         allowed_exts = settings.ALLOWED_EXTENSIONS
-        filename = request.path.replace('/download/', '', 1)
-        dwd_file = os.path.join(root, filename)
-        # Security Checks
-        if '../' in filename or not is_safe_path(root, dwd_file):
-            msg = 'Path Traversal Attack Detected'
-            return print_n_send_error_response(request, msg)
-        ext = os.path.splitext(filename)[1]
-        if ext in allowed_exts:
-            if os.path.isfile(dwd_file):
-                wrapper = FileWrapper(
-                    open(dwd_file, 'rb'))  # lgtm [py/path-injection]
-                response = HttpResponse(
-                    wrapper, content_type=allowed_exts[ext])
-                response['Content-Length'] = os.path.getsize(dwd_file)
-                return response
-        if filename.endswith(('screen/screen.png', '-icon.png')):
-            return HttpResponse('')
-    return HttpResponse(status=404)
+        if not is_md5(checksum):
+            return HttpResponse(
+                'Invalid MD5 Hash',
+                status=HTTP_STATUS_404)
+        robj = RecentScansDB.objects.filter(MD5=checksum).first()
+        if not robj:
+            return HttpResponse(
+                'Scan hash not found',
+                status=HTTP_STATUS_404)
+        file_ext = f'.{robj.SCAN_TYPE}'
+        if file_ext not in allowed_exts.keys():
+            return HttpResponse(
+                'Invalid Scan Type',
+                status=HTTP_STATUS_404)
+        filename = f'{checksum}{file_ext}'
+        dwd_file = Path(settings.UPLD_DIR) / checksum / filename
+        if not dwd_file.exists():
+            return HttpResponse(
+                'File not found',
+                status=HTTP_STATUS_404)
+        return file_download(
+            dwd_file,
+            filename,
+            allowed_exts[file_ext])
+    except Exception:
+        logger.exception('Download Binary Failed')
+        return HttpResponse(
+            'Failed to download file due to an error',
+            status=HTTP_SERVER_ERROR)
+
+
+@login_required
+@require_http_methods(['GET'])
+def download(request):
+    """Download from mobsf downloads directory."""
+    root = settings.DWD_DIR
+    filename = request.path.replace('/download/', '', 1)
+    dwd_file = Path(root) / filename
+
+    # Security Checks
+    if '../' in filename or not is_safe_path(root, dwd_file):
+        msg = 'Path Traversal Attack Detected'
+        return print_n_send_error_response(request, msg)
+
+    # File and Extension Check
+    ext = dwd_file.suffix
+    allowed_exts = settings.ALLOWED_EXTENSIONS
+    if ext in allowed_exts and dwd_file.is_file():
+        return file_download(
+            dwd_file,
+            None,
+            allowed_exts[ext])
+
+    # Special Case for Certain Image Files
+    if filename.endswith(('screen/screen.png', '-icon.png')):
+        return HttpResponse('')
+
+    return HttpResponse(status=HTTP_STATUS_404)
 
 
 @login_required

mobsf/StaticAnalyzer/views/common/firebase.py

@@ -29,56 +29,51 @@
         'severity': INFO,
         'description': 'The app talks to Firebase database at %s',
     },
+    'firebase_db_check_failed': {
+        'title': 'Firebase DB check failed',
+        'severity': INFO,
+        'description': (
+            'Failed to check Firebase DB URL. Error: %s'),
+    },
     'firebase_remote_config_enabled': {
         'title': 'Firebase Remote Config enabled',
         'severity': WARNING,
         'description': (
             'The Firebase Remote Config at %s is enabled.'
             ' Ensure that the configurations are not sensitive.'
-            '\nThis is indicated by the response:\n\n%s'),
+            ' This is indicated by the response: %s'),
     },
     'firebase_remote_config_disabled': {
         'title': 'Firebase Remote Config disabled',
         'severity': SECURE,
         'description': (
             'Firebase Remote Config is disabled for %s.'
-            '\nThis is indicated by the response:\n\n%s'),
+            ' This is indicated by the response: %s'),
     },
     'firebase_remote_config_failed': {
         'title': 'Firebase Remote Config check failed',
+        'severity': INFO,
         'description': (
             'Failed to check for Firebase Remote Config.'
-            ' Please verify this manually. Error:\n\n%s'),
-        'severity': INFO,
+            ' Please verify this manually. Error: %s'),
     },
 }
 
 
 def firebase_analysis(checksum, code_an_dic):
     """Firebase Analysis."""
-    urls = list(set(code_an_dic['urls_list']))
     findings = []
-    finds = None
     logger.info('Starting Firebase Analysis')
     # Check for Firebase Database
-    logger.info('Looking for Firebase URL(s)')
-    for url in urls:
-        if 'firebaseio.com' not in url:
-            continue
-        returl, is_open = open_firebase(checksum, url)
-        if is_open:
-            item = FIREBASE_FINDINGS['firebase_db_open']
-            item['description'] = item['description'] % returl
-            findings.append(item)
-        else:
-            item = FIREBASE_FINDINGS['firebase_db_exists']
-            item['description'] = item['description'] % returl
-            findings.append(item)
+    db_finds = firebase_db_check(
+        checksum, code_an_dic)
+    if db_finds:
+        findings.extend(db_finds)
     # Check for Firebase Remote Config
-    firebase_creds = code_an_dic.get('firebase_creds')
-    finds = firebase_remote_config(checksum, firebase_creds)
-    if finds:
-        findings.extend(finds)
+    config_finds = firebase_remote_config(
+        checksum, code_an_dic)
+    if config_finds:
+        findings.extend(config_finds)
     return findings
 
 
@@ -93,15 +88,16 @@ def open_firebase(checksum, url):
         if not purl.netloc.endswith('firebaseio.com'):
             logger.warning(invalid)
             return url, False
-        base_url = '{}://{}/.json'.format(purl.scheme, purl.netloc)
+        base_url = f'{purl.scheme}://{purl.netloc}/.json'
         proxies, verify = upstream_proxy('https')
         headers = {
             'User-Agent': ('Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1)'
                            ' AppleWebKit/537.36 (KHTML, like Gecko) '
                            'Chrome/39.0.2171.95 Safari/537.36')}
-        resp = requests.get(base_url, headers=headers,
-                            proxies=proxies, verify=verify,
-                            allow_redirects=False)
+        resp = requests.get(
+            base_url, headers=headers,
+            proxies=proxies, verify=verify,
+            allow_redirects=False)
         if resp.status_code == 200:
             return base_url, True
     except Exception as exp:
@@ -111,11 +107,48 @@ def open_firebase(checksum, url):
     return url, False
 
 
-def firebase_remote_config(checksum, creds):
+def firebase_db_check(checksum, code_an_dic):
+    logger.info('Looking for Firebase URL(s)')
+    findings = []
+    try:
+        urls = list(set(code_an_dic['urls_list']))
+        for url in urls:
+            if 'firebaseio.com' not in url:
+                continue
+            returl, is_open = open_firebase(checksum, url)
+            if is_open:
+                rule = FIREBASE_FINDINGS['firebase_db_open']
+                findings.append({
+                    'title': rule['title'],
+                    'severity': rule['severity'],
+                    'description': rule['description'] % returl,
+                })
+            else:
+                rule = FIREBASE_FINDINGS['firebase_db_exists']
+                findings.append({
+                    'title': rule['title'],
+                    'severity': rule['severity'],
+                    'description': rule['description'] % returl,
+                })
+    except Exception as exp:
+        msg = 'Failed to check for Firebase DB URL'
+        logger.warning(msg)
+        append_scan_status(checksum, msg, repr(exp))
+        rule = FIREBASE_FINDINGS['firebase_db_check_failed']
+        findings.append({
+            'title': rule['title'],
+            'severity': rule['severity'],
+            'description': rule['description'] % repr(exp),
+        })
+    return findings
+
+
+def firebase_remote_config(checksum, code_an_dic):
     """Check for Firebase Remote Config."""
     url = None
     findings = []
     try:
+        creds = code_an_dic.get('firebase_creds')
         if not creds:
             return None
         google_api_key = creds.get('google_api_key')
@@ -152,23 +185,36 @@ def firebase_remote_config(checksum, creds):
         if response.status_code == 200:
             resp = response.json()
             if resp.get('state') == 'NO_TEMPLATE':
-                item = FIREBASE_FINDINGS['firebase_remote_config_disabled']
-                item['description'] = item['description'] % (url, resp)
-                findings.append(item)
+                rule = FIREBASE_FINDINGS['firebase_remote_config_disabled']
+                findings.append({
+                    'title': rule['title'],
+                    'severity': rule['severity'],
+                    'description': rule['description'] % (url, resp),
+                })
             else:
-                item = FIREBASE_FINDINGS['firebase_remote_config_enabled']
-                item['description'] = item['description'] % (url, resp)
-                findings.append(item)
+                rule = FIREBASE_FINDINGS['firebase_remote_config_enabled']
+                findings.append({
+                    'title': rule['title'],
+                    'severity': rule['severity'],
+                    'description': rule['description'] % (url, resp),
+                })
         else:
-            item = FIREBASE_FINDINGS['firebase_remote_config_disabled']
+            rule = FIREBASE_FINDINGS['firebase_remote_config_disabled']
             response_msg = f'The response code is {response.status_code}'
-            item['description'] = item['description'] % (url, response_msg)
-            findings.append(item)
+            findings.append({
+                'title': rule['title'],
+                'severity': rule['severity'],
+                'description': rule['description'] % (url, response_msg),
+            })
     except Exception as exp:
         msg = 'Failed to check for Firebase Remote Config'
         logger.warning(msg)
         append_scan_status(checksum, msg, repr(exp))
-        item = FIREBASE_FINDINGS['firebase_remote_config_failed']
-        item['description'] = item['description'] % repr(exp)
-        findings.append(item)
+
+        rule = FIREBASE_FINDINGS['firebase_remote_config_failed']
+        findings.append({
+            'title': rule['title'],
+            'severity': rule['severity'],
+            'description': rule['description'] % repr(exp),
+        })
     return findings

mobsf/templates/general/recent.html

@@ -102,6 +102,7 @@ <h3 class="box-title"><i class="fa fa-rocket"></i> Recent Scans</h3>
                             <td><p>
                                    <a class="btn btn-outline-primary btn-sm" href="{% url "pdf" checksum=e.MD5%}"><i class="fas fa-file-pdf"></i></a>
                                    <a class="btn btn-outline-info btn-sm" href="../../../{{ e.ANALYZER }}/{{e.MD5}}/?rescan=1"><i class="fas fa-sync-alt"></i></a>
+                                   <a class="btn btn-outline-secondary btn-sm" href="{% url "download_binary" checksum=e.MD5%}"><i class="fas fa-download"></i></a>
                                 </p>
                             {% if '.apk' == e.FILE_NAME|slice:"-4:" or '.xapk' == e.FILE_NAME|slice:"-5:" or '.apks' == e.FILE_NAME|slice:"-5:" or '.aab' == e.FILE_NAME|slice:"-4:"%}
                                 <p><a class="diffButton btn btn-warning btn-sm" id="{{ e.MD5 }}_{{ e.FILE_NAME }}"><i class="fas fa-not-equal"></i> Diff or Compare</a>

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "mobsf"
-version = "4.1.3"
+version = "4.1.4"
 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]>"]