Add support for pulling split apks, Fixes #2271

mobsf/DynamicAnalyzer/views/android/dynamic_analyzer.py

@@ -325,16 +325,19 @@ def trigger_static_analysis(request, checksum):
             err = 'Cannot connect to Android Runtime'
             return print_n_send_error_response(request, err)
         env = Environment(identifier)
-        apk_file = env.get_apk(checksum, package)
-        if not apk_file:
+        scan_type = env.get_apk(checksum, package)
+        if not scan_type:
             err = 'Failed to download APK file'
             return print_n_send_error_response(request, err)
+        file_name = f'{package}.apk'
+        if scan_type == 'apks':
+            file_name = f'{file_name}s'
         data = {
             'analyzer': 'static_analyzer',
             'status': 'success',
             'hash': checksum,
-            'scan_type': 'apk',
-            'file_name': f'{package}.apk',
+            'scan_type': scan_type,
+            'file_name': file_name,
         }
         add_to_recent_scan(data)
         return HttpResponseRedirect(f'/static_analyzer/{checksum}/')

mobsf/DynamicAnalyzer/views/android/environment.py

@@ -7,6 +7,7 @@
 import tempfile
 import threading
 import time
+from pathlib import Path
 from base64 import b64encode
 from hashlib import md5
 
@@ -436,31 +437,71 @@
             device_packages[md5] = (pkg, apk)
         return device_packages
 
-    def get_apk(self, checksum, package):
-        """Download APK from device."""
-        try:
-            out_dir = os.path.join(settings.UPLD_DIR, checksum + '/')
-            if not os.path.exists(out_dir):
-                os.makedirs(out_dir)
-            out_file = os.path.join(out_dir, f'{checksum}.apk')
-            if is_file_exists(out_file):
-                return out_file
+    def download_apk_packages(self, pkg_path, out_file):
+        """Download APK package(s)."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            # Download APK package(s)
+            # Can be single or multiple packages
             out = self.adb_command([
-                'pm',
-                'path',
-                package], True)
-            out = out.decode('utf-8').rstrip()
-            path = out.split('package:', 1)[1].strip()
-            logger.info('Downloading APK')
-            self.adb_command([
                 'pull',
-                path,
-                out_file,
+                pkg_path.as_posix(),
+                temp_dir,
             ])
-            if is_file_exists(out_file):
-                return out_file
+            fmt = out.decode('utf-8').strip()
+            logger.info('ADB Pull Output: %s', fmt)
+            # Filter for APK files in the directory
+            apk_files = []
+            for f in Path(temp_dir).glob('*.apk'):
+                if f.is_file():
+                    apk_files.append(f)
+            # Check if there is exactly one APK file
+            if len(apk_files) == 1:
+                shutil.move(apk_files[0], out_file)
+                return 'apk'
+            else:
+                # If there are multiple APK files, zip them
+                shutil.make_archive(out_file, 'zip', root_dir=temp_dir, base_dir='.')
+                # Rename the zip file to APK
+                apks_file = out_file.with_suffix('.apk')
+                os.rename(out_file.as_posix() + '.zip', apks_file)
+                return 'apks'
+
+    def get_apk_packages(self, package):
+        """Get all APK packages from device."""
+        out = self.adb_command([
+            'pm',
+            'path',
+            package], True)
+        return out.decode('utf-8').strip()
+
+    def get_apk_parent_directory(self, package):
+        """Get parent directory of APK packages."""
+        package_out = self.get_apk_packages(package)
+        package_out = package_out.split()
+        if ('package:' in package_out[0]
+                and package_out[0].endswith('.apk')):
+            path = package_out[0].split('package:', 1)[1].strip()
+            return Path(path).parent
+        return False
+
+    def get_apk(self, checksum, package):
+        """Download APK from device."""
+        try:
+            # Do not download if already exists
+            out_dir = Path(settings.UPLD_DIR) / checksum
+            out_dir.mkdir(parents=True, exist_ok=True)
+            out_file = out_dir / f'{checksum}.apk'
+            if out_file.exists():
+                return 'apk'
+            # Get APK package parent directory
+            pkg_path = self.get_apk_parent_directory(package)
+            if pkg_path:
+                # Download APK package(s)
+                logger.info('Downloading APK')
+                return self.download_apk_packages(pkg_path, out_file)
         except Exception:
-            return False
+            logger.exception('Failed to download APK')
+        return False
 
     def system_check(self, runtime):
         """Check if /system is writable."""

mobsf/MobSF/init.py

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

mobsf/StaticAnalyzer/views/android/xapk.py

@@ -62,7 +62,10 @@
     for apk in unzip(checksum, apks.as_posix(), app_dic['app_dir']):
         full_path = app_dic['app_dir'] / apk
         safe_path = is_safe_path(app_dic['app_dir'], full_path)
-        if (not apk.startswith('config.')
+        if apk.endswith('base.apk') and safe_path:
+            move(full_path, apks)
+            return True
+        if ('config.' not in apk.lower()
                 and apk.endswith('.apk')
                 and safe_path):
             move(full_path, apks)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "mobsf"
-version = "4.1.5"
+version = "4.1.6"
 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 <ajin@opensecurity.in>"]