New module: jadx

bbot/modules/apkpure.py

@@ -1,3 +1,4 @@
+import re
 from pathlib import Path
 from bbot.modules.base import BaseModule
 
@@ -45,9 +46,18 @@ async def download_apk(self, app_id):
         path = None
         url = f"https://d.apkpure.com/b/XAPK/{app_id}?version=latest"
         self.helpers.mkdir(self.output_dir / app_id)
-        file_destination = self.output_dir / app_id / f"{app_id}.xapk"
-        result = await self.helpers.download(url, warn=False, filename=file_destination)
-        if result:
-            self.info(f'Downloaded "{app_id}" from "{url}", saved to {file_destination}')
-            path = file_destination
+        response = await self.helpers.request(url, allow_redirects=True)
+        if response:
+            attachment = response.headers.get("Content-Disposition", "")
+            if "filename" in attachment:
+                match = re.search(r'filename="?([^"]+)"?', attachment)
+                if match:
+                    filename = match.group(1)
+                    extension = filename.split(".")[-1]
+                    content = response.content
+                    file_destination = self.output_dir / app_id / f"{app_id}.{extension}"
+                    with open(file_destination, "wb") as f:
+                        f.write(content)
+                    self.info(f'Downloaded "{app_id}" from "{url}", saved to {file_destination}')
+                    path = file_destination
         return path

bbot/modules/jadx.py

@@ -0,0 +1,111 @@
+from pathlib import Path
+from subprocess import CalledProcessError
+from bbot.modules.internal.base import BaseModule
+
+
+class jadx(BaseModule):
+    watched_events = ["FILESYSTEM"]
+    produced_events = ["FILESYSTEM"]
+    flags = ["passive", "safe"]
+    meta = {
+        "description": "Decompile APKs and XAPKs using JADX",
+        "created_date": "2024-11-04",
+        "author": "@domwhewell-sage",
+    }
+    options = {
+        "threads": 4,
+    }
+    options_desc = {
+        "threads": "Maximum jadx threads for extracting apk's, default: 4",
+    }
+    deps_ansible = [
+        {
+            "name": "Install latest JRE (Debian)",
+            "package": {"name": ["default-jre"], "state": "present"},
+            "become": True,
+            "when": "ansible_facts['os_family'] == 'Debian'",
+        },
+        {
+            "name": "Install latest JRE (Arch)",
+            "package": {"name": ["jre-openjdk"], "state": "present"},
+            "become": True,
+            "when": "ansible_facts['os_family'] == 'Archlinux'",
+        },
+        {
+            "name": "Install latest JRE (Fedora)",
+            "package": {"name": ["which", "java-latest-openjdk-headless"], "state": "present"},
+            "become": True,
+            "when": "ansible_facts['os_family'] == 'RedHat'",
+        },
+        {
+            "name": "Install latest JRE (Alpine)",
+            "package": {"name": ["openjdk11"], "state": "present"},
+            "become": True,
+            "when": "ansible_facts['os_family'] == 'Alpine'",
+        },
+        {
+            "name": "Create jadx directory",
+            "file": {"path": "#{BBOT_TOOLS}/jadx", "state": "directory", "mode": "0755"},
+        },
+        {
+            "name": "Download jadx",
+            "unarchive": {
+                "src": "https://github.com/skylot/jadx/releases/download/v1.5.0/jadx-1.5.0.zip",
+                "include": ["lib/jadx-1.5.0-all.jar", "bin/jadx"],
+                "dest": "#{BBOT_TOOLS}/jadx",
+                "remote_src": True,
+            },
+        },
+    ]
+
+    allowed_file_types = ["java archive", "android application package"]
+
+    async def setup(self):
+        self.threads = self.config.get("threads", 4)
+        return True
+
+    async def filter_event(self, event):
+        if "file" in event.tags:
+            if not event.data["magic_description"].lower() in self.allowed_file_types:
+                return False, f"Jadx is not able to decompile this file type: {event.data['magic_description']}"
+        else:
+            return False, "Event is not a file"
+        return True
+
+    async def handle_event(self, event):
+        path = Path(event.data["path"])
+        output_dir = path.parent / path.name.replace(".", "_")
+        self.helpers.mkdir(output_dir)
+        success = await self.decompile_apk(path, output_dir)
+
+        # If jadx was able to decompile the java archive, emit an event
+        if success:
+            await self.emit_event(
+                {"path": str(output_dir)},
+                "FILESYSTEM",
+                tags="folder",
+                parent=event,
+                context=f'extracted "{path}" to: {output_dir}',
+            )
+        else:
+            output_dir.rmdir()
+
+    async def decompile_apk(self, path, output_dir):
+        command = [
+            f"{self.scan.helpers.tools_dir}/jadx/bin/jadx",
+            "--threads-count",
+            self.threads,
+            "--output-dir",
+            str(output_dir),
+            str(path),
+        ]
+        try:
+            output = await self.run_process(command, check=True)
+        except CalledProcessError as e:
+            self.warning(f"Error decompiling {path}. STDERR: {repr(e.stderr)}")
+            return False
+        if not Path(output_dir / "resources").exists() and not Path(output_dir / "sources").exists():
+            self.warning(f"JADX was unable to decompile {path}.")
+            self.warning(output)
+            return False
+        return True

bbot/test/test_step_2/module_tests/test_module_apkpure.py

@@ -37,6 +37,10 @@ async def setup_after_prep(self, module_test):
         module_test.httpx_mock.add_response(
             url="https://d.apkpure.com/b/XAPK/com.bbot.test?version=latest",
             content=self.apk_file,
+            headers={
+                "Content-Type": "application/vnd.android.package-archive",
+                "Content-Disposition": "attachment; filename=com.bbot.test.apk",
+            },
         )
 
     def check(self, module_test, events):
@@ -61,9 +65,7 @@ def check(self, module_test, events):
                 and e.data["url"] == "https://play.google.com/store/apps/details?id=com.bbot.test"
             ]
         ), "Failed to find bbot android app"
-        filesystem_event = [
-            e for e in events if e.type == "FILESYSTEM" and "com.bbot.test.xapk" in e.data["path"] and "apk" in e.tags
-        ]
+        filesystem_event = [e for e in events if e.type == "FILESYSTEM" and "com.bbot.test.apk" in e.data["path"]]
         assert 1 == len(filesystem_event), "Failed to download apk"
         file = Path(filesystem_event[0].data["path"])
-        assert file.is_file(), "Destination xapk doesn't exist"
+        assert file.is_file(), "Destination apk doesn't exist"

bbot/test/test_step_2/module_tests/test_module_jadx.py

@@ -0,0 +1,55 @@
+from pathlib import Path
+from bbot.core.helpers.libmagic import get_magic_info
+from bbot.test.test_step_2.module_tests.base import ModuleTestBase, tempapkfile
+
+
+class TestJadx(ModuleTestBase):
+    modules_overrides = ["apkpure", "google_playstore", "speculate", "jadx"]
+    apk_file = tempapkfile()
+
+    async def setup_after_prep(self, module_test):
+        await module_test.mock_dns({"blacklanternsecurity.com": {"A": ["127.0.0.99"]}})
+        module_test.httpx_mock.add_response(
+            url="https://play.google.com/store/search?q=blacklanternsecurity&c=apps",
+            text="""<!DOCTYPE html>
+            <html>
+            <head>
+            <title>"blacklanternsecurity" - Android Apps on Google Play</title>
+            </head>
+            <body>
+            <a href="/store/apps/details?id=com.bbot.test&pcampaignid=dontmatchme&pli=1"/>
+            </body>
+            </html>""",
+        )
+        module_test.httpx_mock.add_response(
+            url="https://play.google.com/store/apps/details?id=com.bbot.test",
+            text="""<!DOCTYPE html>
+            <html>
+            <head>
+            <title>BBOT</title>
+            </head>
+            <body>
+            <meta name="appstore:developer_url" content="https://www.blacklanternsecurity.com">
+            </div>
+            </div>
+            </body>
+            </html>""",
+        )
+        module_test.httpx_mock.add_response(
+            url="https://d.apkpure.com/b/XAPK/com.bbot.test?version=latest",
+            content=self.apk_file,
+            headers={
+                "Content-Type": "application/vnd.android.package-archive",
+                "Content-Disposition": "attachment; filename=com.bbot.test.apk",
+            },
+        )
+
+    def check(self, module_test, events):
+        filesystem_events = [e for e in events if e.type == "FILESYSTEM"]
+        apk_event = [e for e in filesystem_events if "file" in e.tags]
+        extension, mime_type, description, confidence = get_magic_info(apk_event[0].data["path"])
+        assert description == "Android Application Package", f"Downloaded file was detected as {description}"
+        extract_event = [e for e in filesystem_events if "folder" in e.tags]
+        assert 1 == len(extract_event), "Failed to extract apk"
+        extract_path = Path(extract_event[0].data["path"])
+        assert extract_path.is_dir(), "Destination apk doesn't exist"