Deduplicate AndroidQF SMS analysis

.github/workflows/add-issue-to-project.yml

@@ -11,7 +11,7 @@ jobs:
     name: Add issue to project
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/add-to-project@v2
+      - uses: actions/add-to-project@v0.5.0
         with:
           # You can target a project in a different organization
           # to the issue

.github/workflows/mypy.yml

@@ -11,13 +11,13 @@ jobs:
 
     steps:
       - name: Checkout
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
       - name: Setup Python
-        uses: actions/setup-python@v7
+        uses: actions/setup-python@v6
         with:
           python-version: "3.10"
       - name: Install uv
-        uses: astral-sh/setup-uv@v10.0.0
+        uses: astral-sh/setup-uv@v7
         with:
           enable-cache: true
       - name: Install Dependencies

.github/workflows/publish-release-docker.yml

@@ -36,10 +36,10 @@ jobs:
             tag-suffix: "-android"
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
       # Uses the `docker/login-action` action to log in to the Container registry registry using the account and password that will publish the packages. Once published, the packages are scoped to the account defined here.
       - name: Log in to the Container registry
-        uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
+        uses: docker/login-action@65b78e6e13532edd9afa3aa52ac7964289d1a9c1
         with:
           registry: ${{ env.REGISTRY }}
           username: ${{ github.actor }}
@@ -47,7 +47,7 @@ jobs:
       # This step uses [docker/metadata-action](https://github.com/docker/metadata-action#about) to extract tags and labels that will be applied to the specified image. The `id` "meta" allows the output of this step to be referenced in a subsequent step. The `images` value provides the base name for the tags and labels.
       - name: Extract metadata (tags, labels) for Docker
         id: meta
-        uses: docker/metadata-action@v6
+        uses: docker/metadata-action@v5
         with:
           images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
           flavor: |
@@ -61,13 +61,13 @@ jobs:
             type=sha,format=long,suffix=${{ matrix.platform.tag-suffix }}
       # This step sets up some additional capabilities to generate the provenance and sbom attestations
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v4
+        uses: docker/setup-buildx-action@v3
       # This step uses the `docker/build-push-action` action to build the image, based on your repository's `Dockerfile`. If the build succeeds, it pushes the image to GitHub Packages.
       # It uses the `context` parameter to define the build's context as the set of files located in the specified path. For more information, see "[Usage](https://github.com/docker/build-push-action#usage)" in the README of the `docker/build-push-action` repository.
       # It uses the `tags` and `labels` parameters to tag and label the image with the output from the "meta" step.
       - name: Build and push Docker image
         id: push
-        uses: docker/build-push-action@v7
+        uses: docker/build-push-action@v6
         with:
           file: ${{ matrix.platform.dockerfile }}
           context: .
@@ -76,3 +76,4 @@ jobs:
           labels: ${{ steps.meta.outputs.labels }}
           provenance: mode=max
           sbom: true
+

.github/workflows/ruff.yml

@@ -12,13 +12,13 @@ jobs:
 
     steps:
       - name: Checkout
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
       - name: Setup Python
-        uses: actions/setup-python@v7
+        uses: actions/setup-python@v6
         with:
           python-version: "3.10"
       - name: Install uv
-        uses: astral-sh/setup-uv@v10.0.0
+        uses: astral-sh/setup-uv@v7
         with:
           enable-cache: true
       - name: Install Dependencies

.github/workflows/tests.yml

@@ -15,13 +15,13 @@ jobs:
         python-version: ['3.10', '3.11', '3.12', '3.13', '3.14']
 
     steps:
-      - uses: actions/checkout@v7
+      - uses: actions/checkout@v6
       - name: Install uv
-        uses: astral-sh/setup-uv@v10.0.0
+        uses: astral-sh/setup-uv@v7
         with:
           enable-cache: true
       - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v7
+        uses: actions/setup-python@v6
         with:
           python-version: ${{ matrix.python-version }}
       - name: Install Python dependencies

.github/workflows/update-ios-data.yml

@@ -11,12 +11,10 @@ jobs:
   update-ios-version:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v7
-      - uses: actions/setup-python@v7
-      - name: Install script dependencies
-        run: python -m pip install "packaging==26.3"
+      - uses: actions/checkout@v3
+      - uses: actions/setup-python@v4
       - name: Run script to fetch latest iOS releases from Apple RSS feed.
-        run: python .github/workflows/scripts/update-ios-releases.py
+        run: python3 .github/workflows/scripts/update-ios-releases.py
       - name: Create Pull Request
         uses: peter-evans/create-pull-request@v8
         with:

README.md

@@ -82,11 +82,6 @@ Module-running `check-*` commands can load custom Python modules with
 [development documentation](https://docs.mvt.re/en/latest/development/) for
 details.
 
-Users can also add top-level commands to `mvt-ios` and `mvt-android` from
-installed Python packages or local files and folders. See the
-[custom CLI command documentation](https://docs.mvt.re/en/latest/custom_commands/)
-for the plugin entry points and `--load-command` interface.
-
 
 ## License
 

docs/android/backup.md

@@ -57,9 +57,3 @@ If the backup is encrypted, ABE will prompt you to enter the password.
 Alternatively, [ab-decrypt](https://github.com/joernheissler/ab-decrypt) can be used for that purpose.
 
 You can then extract SMSs with MVT by passing the folder path as parameter instead of the `.ab` file: `mvt-android check-backup --output /path/to/results/ /path/to/backup/` (the path to backup given should be the folder containing the `apps` folder).
-
-When an output folder is specified, URLs extracted from SMS and MMS messages
-are also written to `urls.json`. Each entry contains the URL, its expanded
-destination when MVT resolved a shortened URL during indicator checking, the
-message timestamp, and the `sms` source. The same file is created by
-`check-androidqf` when its nested Android backup contains messages with URLs.

docs/custom_commands.md

@@ -1,107 +0,0 @@
-# Custom CLI Commands
-
-MVT can load additional top-level commands into `mvt-ios` and `mvt-android`.
-Custom commands are different from [custom forensic modules](development.md#custom-modules):
-commands add new CLI operations, while modules add analysis steps to existing
-`check-*` commands.
-
-!!! warning
-
-    Custom commands run as trusted Python code inside the MVT process. Install
-    or load commands only from sources you trust. MVT does not sandbox
-    third-party commands, and the MVT maintainers do not maintain them.
-
-## Install a Command Package
-
-Python packages can register a Click command or group for either MVT CLI. A
-minimal package can expose this command from `my_mvt_plugin.py`:
-
-```python
-import click
-
-
-@click.command()
-@click.argument("path", type=click.Path(exists=True))
-def summarize(path):
-    """Summarize an acquisition."""
-    click.echo(f"Summarizing {path}")
-```
-
-Register the object in the package's `pyproject.toml`. The entry-point name is
-the command users invoke:
-
-```toml
-[project.entry-points."mvt.ios.cli_plugins"]
-summarize = "my_mvt_plugin:summarize"
-
-[project.entry-points."mvt.android.cli_plugins"]
-summarize = "my_mvt_plugin:summarize"
-```
-
-Use only the iOS or Android group if the command is platform-specific. After
-installing the package in the same environment as MVT, it appears directly in
-the appropriate CLI:
-
-```bash
-mvt-ios summarize ./ios-backup
-mvt-android summarize ./androidqf-output
-```
-
-For a `pipx` installation of MVT, inject the plugin into MVT's environment:
-
-```bash
-pipx inject mvt my-mvt-plugin
-```
-
-When MVT is installed in an active virtual environment, install the plugin with
-`pip` in that environment.
-
-## Load a Command File
-
-For local commands that are not packaged, create a Python file that exports one
-Click command or group named `cli`:
-
-```python
-import click
-
-
-@click.command("case-summary")
-@click.argument("path", type=click.Path(exists=True))
-def cli(path):
-    """Summarize a case directory."""
-    click.echo(f"Summarizing {path}")
-```
-
-Pass the file before the custom command name:
-
-```bash
-mvt-ios --load-command ./case_summary.py case-summary ./ios-backup
-```
-
-`--load-command` can be repeated and also accepts a folder. MVT loads
-non-hidden top-level `*.py` files in sorted order and skips `__init__.py`.
-Every loaded file must export one `cli` object.
-
-To load a file or folder on every invocation, set the platform-specific
-environment variable:
-
-```bash
-export MVT_IOS_CUSTOM_COMMANDS=./ios_commands
-export MVT_ANDROID_CUSTOM_COMMANDS=./android_commands
-```
-
-## Naming and Errors
-
-Built-in MVT commands cannot be replaced. External command names must also be
-unique; when installed packages or environment paths collide, MVT keeps the
-first command and logs a warning. A collision from an explicit
-`--load-command` is a usage error.
-
-A package entry point or environment command that cannot be imported appears
-as a marked broken command without preventing other MVT commands from working.
-Invoke that command to see its package or file source and the underlying error.
-An invalid command supplied explicitly with `--load-command` fails immediately
-with a usage error.
-
-Installed command packages use the entry-point name as the CLI command name.
-The entry point must resolve to a `click.Command` or `click.Group`.

docs/development.md

@@ -82,16 +82,14 @@ class ExampleCustomModule(MVTModule):
         return None
 ```
 
-Use `supported_commands` to declare the platform/command pairs a module
-supports. Empty `supported_commands` means the module will not run and MVT logs
-a warning. This explicit declaration is required for every command. Supported
-pairs are:
+Use `supported_commands` to restrict a module to specific platform/command
+pairs. Missing or empty `supported_commands` means the module is available to
+all commands, which keeps older modules compatible. Supported pairs are:
 
 ```python
 ("ios", "check-backup")
 ("ios", "check-fs")
 ("ios", "check-iocs")
-("ios", "check-sysdiagnose")
 ("android", "check-backup")
 ("android", "check-bugreport")
 ("android", "check-androidqf")

docs/iocs.md

@@ -37,10 +37,8 @@ export MVT_STIX2="/home/user/IOC1.stix2:/home/user/IOC2.stix2"
 ## Network Access
 
 When checking URL indicators, MVT follows recognized shortened URLs with an
-HTTP `HEAD` request. URL checks are deduplicated and run concurrently, with at
-most 20 requests in progress at a time. Redirects within an individual URL
-chain are still followed sequentially. The following environment variables
-control these requests:
+HTTP `HEAD` request. The following environment variables control these
+requests:
 
 - `MVT_NETWORK_ACCESS_ALLOWED` enables or disables network requests. It defaults
   to `true`. Set it to `false` to prevent MVT from attempting to resolve
@@ -75,3 +73,4 @@ You can automaticallly download the latest public indicator files with the comma
 
 Please [open an issue](https://github.com/mvt-project/mvt/issues/) to suggest new sources of STIX-formatted IOCs.
 
+

docs/ios/records.md

@@ -312,15 +312,6 @@ If indicators are provided through the command-line, they are checked against th
 
 ---
 
-### `urls.json`
-
-This JSON file collects URLs extracted from SMS, iMessage, and WhatsApp
-messages. Each entry contains the original URL, its expanded destination when
-MVT resolved a shortened URL during indicator checking, the message timestamp,
-and its `sms` or `whatsapp` source.
-
----
-
 ### `sms_attachments.json`
 
 !!! info "Availability"

docs/ios/sysdiagnose.md

@@ -1,50 +0,0 @@
-# Check an iOS Sysdiagnose
-
-`mvt-ios check-sysdiagnose` prepares an iOS sysdiagnose archive for analysis by
-custom MVT modules. MVT does not include built-in sysdiagnose modules. You must
-load at least one custom module that explicitly supports this command.
-
-The command accepts either an extracted sysdiagnose directory or the original
-gzip-compressed tar archive.
-
-```bash
-mvt-ios check-sysdiagnose \
-    --load-module ./sysdiagnose_modules.py \
-    --output ./results \
-    ./sysdiagnose_2024.01.02_03-04-05+0200.tar.gz
-```
-
-Use `--hashes` to include hashes for analyzed files in `info.json`, and
-`--list-modules` to display the eligible custom modules without running them.
-
-## Writing a custom module
-
-Extend `SysdiagnoseExtraction` to access the archive contents consistently for
-both directory and tar inputs. Each module must declare the command explicitly
-in `supported_commands`.
-
-```python
-from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
-
-
-class ExampleSysdiagnoseModule(SysdiagnoseExtraction):
-    supported_commands = (("ios", "check-sysdiagnose"),)
-    slug = "example_sysdiagnose"
-
-    def run(self):
-        paths = self._get_files_by_pattern("*/example.log")
-        if paths:
-            content = self._get_file_content(paths[0]).decode("utf-8", "replace")
-            self.results = [{"content": content}]
-
-    def check_indicators(self):
-        pass
-
-    def serialize(self, result):
-        return None
-```
-
-The base class provides `from_sysdiagnose_folder()` and
-`from_sysdiagnose_tar()` setup hooks, as well as protected file lookup, file
-reading, and timezone extraction helpers. IPS crash-report metadata is exposed
-on `ips_files`.

mkdocs.yml

@@ -3,8 +3,6 @@ repo_url: https://github.com/mvt-project/mvt
 edit_uri: edit/main/docs/
 copyright: Copyright © 2021-2023 MVT Project Developers
 site_description: Mobile Verification Toolkit Documentation
-not_in_nav: |
-    android/adb.md
 markdown_extensions:
     - attr_list
     - admonition
@@ -30,8 +28,6 @@ nav:
     - Welcome: "index.md"
     - Introduction: "introduction.md"
     - Installation: "install.md"
-    - Command Completion: "command_completion.md"
-    - Custom CLI Commands: "custom_commands.md"
     - Using Docker: "docker.md"
     - MVT for iOS:
         - iOS Forensic Methodology: "ios/methodology.md"
@@ -43,7 +39,6 @@ nav:
         - Check a Filesystem Dump:
             - Dumping the filesystem: "ios/filesystem/dump.md"
             - Check a Filesystem Dump with mvt-ios: "ios/filesystem/check.md"
-        - Check a Sysdiagnose: "ios/sysdiagnose.md"
         - Records extracted by mvt-ios: "ios/records.md"
     - MVT for Android:
         - Android Forensic Methodology: "android/methodology.md"

pyproject.toml

@@ -17,25 +17,25 @@ classifiers = [
     "Programming Language :: Python",
 ]
 dependencies = [
-    "click==8.4.2",
-    "rich==15.0.0",
+    "click==8.3.3",
+    "rich==14.3.3",
     "tld==0.13.2",
-    "requests==2.34.2",
-    "simplejson==4.1.1",
-    "packaging==26.3",
+    "requests==2.33.1",
+    "simplejson==3.20.2",
+    "packaging==26.2",
     "appdirs==1.4.4",
     "iOSbackup==0.9.925",
     "adb-shell[usb]==0.4.4",
-    "libusb1==3.4.0",
-    "cryptography==50.0.0",
+    "libusb1==3.3.1",
+    "cryptography==47.0.0",
     "PyYAML>=6.0.2",
     "pyahocorasick==2.3.1",
-    "betterproto2==0.10.0",
-    "pydantic==2.13.4",
-    "pydantic-settings==2.15.0",
+    "betterproto2==0.9.1",
+    "pydantic==2.13.3",
+    "pydantic-settings==2.14.0",
     "NSKeyedUnArchiver==1.5.2",
     "python-dateutil==2.9.0.post0",
-    "tzdata==2026.3",
+    "tzdata==2026.2",
 ]
 requires-python = ">= 3.10"
 
@@ -65,9 +65,9 @@ dev = [
 docs = [
     "mkdocs==1.6.1",
     "mkdocs-autorefs==1.4.4",
-    "mkdocs-material==9.7.7",
+    "mkdocs-material==9.7.6",
     "mkdocs-material-extensions==1.3.1",
-    "mkdocstrings==1.0.6",
+    "mkdocstrings==1.0.4",
 ]
 
 [build-system]

src/mvt/android/artifacts/dumpsys_adb.py

@@ -13,13 +13,6 @@ from .artifact import AndroidArtifact
 class DumpsysADBArtifact(AndroidArtifact):
     multiline_fields = ["user_keys", "keystore"]
 
-    @staticmethod
-    def is_structural_line(key: str, vals: list) -> bool:
-        if key == "}":
-            return True
-        # XML keystore continuations also split on "=", but never into an identifier.
-        return len(vals) == 2 and key.isidentifier()
-
     def indented_dump_parser(self, dump_data):
         """
         Parse the indented dumpsys output, generated by DualDumpOutputStream in Android.
@@ -48,18 +41,10 @@ class DumpsysADBArtifact(AndroidArtifact):
                 if key == "":
                     # If the line is empty, it's the terminator for the multiline value
                     in_multiline = False
-                    if isinstance(stack[-1], list):
-                        stack.pop()
-                    continue
-
-                if not self.is_structural_line(key, vals):
-                    current_dict.append(line.lstrip())
-                    continue
-
-                in_multiline = False
-                if isinstance(stack[-1], list):
                     stack.pop()
-                current_dict = stack[-1]
+                else:
+                    current_dict.append(line.lstrip())
+                continue
 
             if key == "}":
                 stack.pop()

src/mvt/android/artifacts/dumpsys_battery_history.py

@@ -31,38 +31,21 @@ class DumpsysBatteryHistoryArtifact(AndroidArtifact):
             if line.strip() == "":
                 break
 
-            time_parts = line.strip().split()
-            time_elapsed = time_parts[0]
-            if (
-                len(time_parts) > 1
-                and len(time_parts[0]) == 5
-                and time_parts[0][2] == "-"
-                and ":" in time_parts[1]
-            ):
-                time_elapsed = " ".join(time_parts[:2])
+            time_elapsed = line.strip().split(" ", 1)[0]
 
             event = ""
             if line.find("+job") > 0:
                 event = "start_job"
-                payload = line.split("+job=", 1)[1]
-                uid, separator, service = payload.partition(":")
-                if not separator:
-                    continue
-                service = service.strip().strip('"')
+                uid = line[line.find("+job") + 5 : line.find(":")]
+                service = line[line.find(":") + 1 :].strip('"')
                 package_name = service.split("/")[0]
             elif line.find("-job") > 0:
                 event = "end_job"
-                payload = line.split("-job=", 1)[1]
-                uid, separator, service = payload.partition(":")
-                if not separator:
-                    continue
-                service = service.strip().strip('"')
+                uid = line[line.find("-job") + 5 : line.find(":")]
+                service = line[line.find(":") + 1 :].strip('"')
                 package_name = service.split("/")[0]
             elif line.find("+running +wake_lock=") > 0:
-                payload = line.split("+running +wake_lock=", 1)[1]
-                uid, separator, _ = payload.partition(":")
-                if not separator:
-                    continue
+                uid = line[line.find("+running +wake_lock=") + 21 : line.find(":")]
                 event = "wake"
                 service = (
                     line[line.find("*walarm*:") + 9 :].split(" ")[0].strip('"').strip()

src/mvt/android/artifacts/dumpsys_dbinfo.py

@@ -29,9 +29,10 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
 
     def parse(self, output: str) -> None:
         rxp = re.compile(
-            r".*\[((?:[0-9]{4}-)?[0-9]{2}-[0-9]{2} "
-            r"[0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\]\s*"
-            r"(?:\[Pid:\((\d+)\)\])?([\w-]+).*?sql=\"(.+?)\""
+            r".*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\].*\[Pid:\((\d+)\)\](\w+).*sql\=\"(.+?)\""
+        )  # pylint: disable=line-too-long
+        rxp_no_pid = re.compile(
+            r".*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\][ ]{1}(\w+).*sql\=\"(.+?)\""
         )  # pylint: disable=line-too-long
 
         pool = None
@@ -55,16 +56,29 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
                 pool = None
                 continue
 
-            match = rxp.match(line)
-            if not match:
-                continue
+            matches = rxp.findall(line)
+            if not matches:
+                matches = rxp_no_pid.findall(line)
+                if not matches:
+                    continue
 
-            result = {
-                "isodate": match.group(1),
-                "action": match.group(3),
-                "sql": match.group(4),
-                "path": pool,
-            }
-            if match.group(2):
-                result["pid"] = match.group(2)
-            self.results.append(result)
+                match = matches[0]
+                self.results.append(
+                    {
+                        "isodate": match[0],
+                        "action": match[1],
+                        "sql": match[2],
+                        "path": pool,
+                    }
+                )
+            else:
+                match = matches[0]
+                self.results.append(
+                    {
+                        "isodate": match[0],
+                        "pid": match[1],
+                        "action": match[2],
+                        "sql": match[3],
+                        "path": pool,
+                    }
+                )

src/mvt/android/artifacts/dumpsys_receivers.py

@@ -96,18 +96,6 @@ class DumpsysReceiversArtifact(AndroidArtifact):
                 self.results[intent] = []
                 continue
 
-            parts = line.strip().split(" ")
-            if len(parts) < 2:
-                # A single-token line here is not a receiver. Real dumpstate
-                # output can print an action header mis-indented (observed with
-                # 15 leading spaces instead of 6), which used to raise
-                # IndexError and abort the whole module. Treat a trailing-colon
-                # token as the next action, skip anything else.
-                if parts[0].endswith(":"):
-                    intent = parts[0][:-1]
-                    self.results.setdefault(intent, [])
-                continue
-
             # If we are not in an intent block yet, skip.
             if not intent:
                 continue
@@ -121,7 +109,7 @@ class DumpsysReceiversArtifact(AndroidArtifact):
 
             # If we got this far, we are processing receivers for the
             # activities we are interested in.
-            receiver = parts[1]
+            receiver = line.strip().split(" ")[1]
             package_name = receiver.split("/")[0]
 
             self.results[intent].append(

src/mvt/android/artifacts/tombstone_crashes.py

@@ -34,7 +34,6 @@ TOMBSTONE_TEXT_KEY_MAPPINGS = {
     "signal": "signal_info",
     "code": "code",
     "Cause": "cause",
-    "Abort message": "abort_message",
 }
 
 
@@ -68,8 +67,6 @@ class TombstoneCrashResult(pydantic.BaseModel):
     uid: int
     signal_info: SignalInfo
     cause: Optional[str] = None
-    causes: Optional[List[dict]] = None
-    abort_message: Optional[str] = None
     extra: Optional[str] = None
 
 
@@ -209,11 +206,6 @@ class TombstoneCrashArtifact(AndroidArtifact):
         return True
 
     def _load_pid_line(self, line: str, tombstone: dict) -> bool:
-        # The first pid line identifies the crashing thread. Full text tombstones
-        # contain additional pid lines for the other threads in the process.
-        if "pid" in tombstone:
-            return True
-
         try:
             parts = line.split(" >>> ") if " >>> " in line else line.split(">>>")
             process_info = parts[0]

src/mvt/android/cli.py

@@ -4,16 +4,9 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-from zipfile import BadZipFile
 
 import click
 
-from mvt.common.cli_plugins import (
-    ANDROID_CLI_PLUGIN_GROUP,
-    MVT_ANDROID_CUSTOM_COMMANDS_ENV,
-    load_cli_commands_option,
-    register_cli_plugins,
-)
 from mvt.common.cmd_check_iocs import CmdCheckIOCS
 from mvt.common.completion import (
     SUPPORTED_SHELLS,
@@ -88,7 +81,6 @@ def _load_custom_modules(load_module):
 # Main
 # ==============================================================================
 @click.group(invoke_without_command=False)
-@load_cli_commands_option
 @click.option(
     "--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
 )
@@ -220,10 +212,7 @@ def check_bugreport(
 
     log.info("Checking Android bug report at path: %s", bugreport_path)
 
-    try:
-        cmd.run()
-    except BadZipFile as exc:
-        raise click.ClickException(f"Invalid bugreport archive: {exc}") from exc
+    cmd.run()
     cmd.show_alerts_brief()
     cmd.show_support_message()
 
@@ -512,10 +501,3 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
 def download_indicators():
     ioc_updates = IndicatorsUpdates()
     ioc_updates.update()
-
-
-register_cli_plugins(
-    cli,
-    entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
-    environment_variable=MVT_ANDROID_CUSTOM_COMMANDS_ENV,
-)

src/mvt/android/cmd_check_androidqf.py

@@ -3,7 +3,6 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-import json
 import logging
 import os
 import shutil
@@ -95,47 +94,6 @@ class CmdAndroidCheckAndroidQF(Command):
             self.__zip = zipfile.ZipFile(self.target_path)
             self.__files = self.__zip.namelist()
 
-        self._load_acquisition_context()
-
-    def _load_acquisition_context(self) -> None:
-        """Pass AndroidQF acquisition metadata to nested commands and modules."""
-        context = {}
-        metadata_files = [
-            file_path
-            for file_path in self.__files
-            if file_path.replace("\\", "/").rsplit("/", 1)[-1] == "acquisition.json"
-        ]
-        for file_path in metadata_files:
-            try:
-                metadata = json.loads(self._get_file_content(file_path))
-                if isinstance(metadata, dict):
-                    context["started"] = metadata.get("started")
-                    context["adb_host_public_key"] = metadata.get("adb_host_public_key")
-                    break
-            except (json.JSONDecodeError, OSError, TypeError, UnicodeDecodeError):
-                self.log.warning(
-                    'Unable to read AndroidQF acquisition metadata "%s"', file_path
-                )
-
-        if not context.get("adb_host_public_key"):
-            key_files = [
-                file_path
-                for file_path in self.__files
-                if file_path.replace("\\", "/").rsplit("/", 1)[-1] == "adb_host_key.pub"
-            ]
-            for file_path in key_files:
-                try:
-                    context["adb_host_public_key"] = self._get_file_content(
-                        file_path
-                    ).decode("utf-8")
-                    break
-                except (OSError, UnicodeDecodeError):
-                    self.log.warning(
-                        'Unable to read AndroidQF ADB host key "%s"', file_path
-                    )
-
-        self.module_options["androidqf_acquisition"] = context
-
     def module_init(self, module: AndroidQFModule) -> None:  # type: ignore[override]
         if self.__format == "zip" and self.__zip:
             module.from_zip(self.__zip, self.__files)
@@ -262,7 +220,6 @@ class CmdAndroidCheckAndroidQF(Command):
             cmd.run()
 
             self.timeline.extend(cmd.timeline)
-            self.url_results.extend(cmd.url_results)
             self.alertstore.extend(cmd.alertstore.alerts)
         finally:
             if bugreport:
@@ -300,7 +257,6 @@ class CmdAndroidCheckAndroidQF(Command):
         cmd.run()
 
         self.timeline.extend(cmd.timeline)
-        self.url_results.extend(cmd.url_results)
         self.alertstore.extend(cmd.alertstore.alerts)
         return True
 
@@ -340,7 +296,6 @@ class CmdAndroidCheckAndroidQF(Command):
 
             elif self.__format == "zip" and self.__zip:
                 temp_dir = tempfile.mkdtemp(prefix="mvt_intrusion_logs_")
-                temp_root = Path(temp_dir).resolve()
                 for entry in intrusion_log_files:
                     normalized = entry.replace("\\", "/")
                     idx = normalized.find("intrusion_logs/")
@@ -348,15 +303,9 @@ class CmdAndroidCheckAndroidQF(Command):
                     if not relative or relative.endswith("/"):
                         continue
 
-                    target = (temp_root / relative).resolve()
-                    if not target.is_relative_to(temp_root):
-                        self.log.warning(
-                            "Skipping unsafe intrusion log archive entry: %s", entry
-                        )
-                        continue
-
-                    target.parent.mkdir(parents=True, exist_ok=True)
-                    with self.__zip.open(entry) as src, target.open("wb") as dst:
+                    target = os.path.join(temp_dir, relative)
+                    os.makedirs(os.path.dirname(target), exist_ok=True)
+                    with self.__zip.open(entry) as src, open(target, "wb") as dst:
                         dst.write(src.read())
 
                 intrusion_logs_path = temp_dir
@@ -380,7 +329,6 @@ class CmdAndroidCheckAndroidQF(Command):
             cmd.run()
 
             self.timeline.extend(cmd.timeline)
-            self.url_results.extend(cmd.url_results)
             self.alertstore.extend(cmd.alertstore.alerts)
             return True
 

src/mvt/android/modules/androidqf/__init__.py

@@ -5,7 +5,6 @@
 
 from .aqf_files import AQFFiles
 from .aqf_getprop import AQFGetProp
-from .aqf_log_timestamps import AQFLogTimestamps
 from .aqf_packages import AQFPackages
 from .aqf_processes import AQFProcesses
 from .aqf_settings import AQFSettings
@@ -18,7 +17,6 @@ ANDROIDQF_MODULES = [
     AQFGetProp,
     AQFSettings,
     AQFFiles,
-    AQFLogTimestamps,
     RootBinaries,
     Mounts,
 ]

src/mvt/android/modules/backup/sms.py

@@ -36,8 +36,6 @@ class SMS(BackupModule):
         if not self.indicators:
             return
 
-        messages = []
-        url_batches = []
         for message in self.results:
             if "body" not in message:
                 continue
@@ -46,21 +44,12 @@ class SMS(BackupModule):
             if message_links == []:
                 message_links = check_for_links(message.get("text", ""))
 
-            messages.append(message)
-            url_batches.append(message_links)
-
-        for message, ioc_match in zip(
-            messages, self.indicators.check_url_batches(url_batches)
-        ):
+            ioc_match = self.indicators.check_urls(message_links)
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", message, matched_indicator=ioc_match.ioc
                 )
-
-    def collect_url_results(self) -> None:
-        for message in self.results:
-            for url in message.get("links", []):
-                self.add_url_result(url, message.get("isodate"), "sms")
+                continue
 
     def run(self) -> None:
         sms_path = "apps/com.android.providers.telephony/d_f/*_sms_backup"

src/mvt/android/modules/bugreport/dumpsys_adb_state.py

@@ -3,15 +3,11 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-import base64
-import binascii
-import datetime
 import logging
 from typing import Optional
 
 from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact
 from mvt.common.module_types import ModuleResults
-from mvt.common.utils import convert_datetime_to_iso
 
 from .base import BugReportModule
 
@@ -57,122 +53,3 @@ class DumpsysADBState(DumpsysADBArtifact, BugReportModule):
                 "Identified a total of %d trusted ADB keys",
                 len(self.results[0].get("user_keys", [])),
             )
-
-    @staticmethod
-    def _key_material(public_key: object) -> str:
-        if isinstance(public_key, bytes):
-            public_key = public_key.decode("utf-8", errors="replace")
-        if not isinstance(public_key, str):
-            return ""
-        return public_key.strip().split(" ", 1)[0]
-
-    @staticmethod
-    def _is_valid_key(public_key: str) -> bool:
-        if not public_key:
-            return False
-        try:
-            return bool(base64.b64decode(public_key, validate=True))
-        except (binascii.Error, ValueError):
-            return False
-
-    @staticmethod
-    def _parse_acquisition_time(value: object) -> Optional[datetime.datetime]:
-        if not isinstance(value, str):
-            return None
-        try:
-            timestamp = datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
-        except ValueError:
-            return None
-        if not timestamp.tzinfo:
-            timestamp = timestamp.replace(tzinfo=datetime.timezone.utc)
-        return timestamp.astimezone(datetime.timezone.utc)
-
-    @staticmethod
-    def _parse_last_connected(value: object) -> Optional[datetime.datetime]:
-        try:
-            return datetime.datetime.fromtimestamp(
-                int(str(value)) / 1000,
-                tz=datetime.timezone.utc,
-            )
-        except (OSError, OverflowError, TypeError, ValueError):
-            return None
-
-    def _trusted_keys(self) -> list[dict]:
-        """Return unique trusted keys, preferring keystore connection metadata."""
-        trusted_keys = []
-        seen = set()
-        for result in self.results:
-            keystore = result.get("keystore", [])
-            candidates = keystore if isinstance(keystore, list) else []
-            candidates = [*candidates, *result.get("user_keys", [])]
-            for candidate in candidates:
-                if not isinstance(candidate, dict):
-                    continue
-                key = self._key_material(candidate.get("key"))
-                identity = key or repr(candidate)
-                if identity in seen:
-                    continue
-                seen.add(identity)
-                trusted_keys.append(candidate)
-        return trusted_keys
-
-    def check_indicators(self) -> None:
-        if "androidqf_acquisition" not in self.module_options:
-            return super().check_indicators()
-
-        context = self.module_options.get("androidqf_acquisition")
-        if not isinstance(context, dict):
-            context = {}
-        acquisition_key = self._key_material(context.get("adb_host_public_key"))
-        if acquisition_key and not self._is_valid_key(acquisition_key):
-            acquisition_key = ""
-        acquisition_time = self._parse_acquisition_time(context.get("started"))
-        cutoff = (
-            acquisition_time - datetime.timedelta(days=1) if acquisition_time else None
-        )
-
-        for trusted_key in self._trusted_keys():
-            key = self._key_material(trusted_key.get("key"))
-            fingerprint = trusted_key.get("fingerprint") or "<unknown key>"
-            user = trusted_key.get("user") or "unknown user"
-            description = f"{fingerprint} ({user})"
-            last_connected = self._parse_last_connected(
-                trusted_key.get("last_connected")
-            )
-            event_time = (
-                convert_datetime_to_iso(last_connected) if last_connected else ""
-            )
-
-            if not self._is_valid_key(key):
-                self.alertstore.low(
-                    f"Found an invalid trusted ADB host key: {description}",
-                    event_time,
-                    trusted_key,
-                )
-                continue
-
-            if not acquisition_key:
-                self.alertstore.low(
-                    "Found a trusted ADB host key, but the AndroidQF acquisition "
-                    f"does not include its host key: {description}",
-                    event_time,
-                    trusted_key,
-                )
-                continue
-
-            if key != acquisition_key:
-                self.alertstore.low(
-                    "Found a trusted ADB host key different from the AndroidQF "
-                    f"acquisition host: {description}",
-                    event_time,
-                    trusted_key,
-                )
-                continue
-
-            if cutoff and last_connected and last_connected <= cutoff:
-                self.alertstore.info(
-                    "Found a trusted ADB host key last connected at least one day "
-                    f"before the AndroidQF acquisition: {description}",
-                    event_time,
-                    trusted_key,
-                )

src/mvt/android/parsers/backup.py

@@ -131,23 +131,20 @@ def decrypt_backup_data(encrypted_backup, password, encryption_algo, format_vers
     if password is None:
         raise InvalidBackupPassword()
 
-    try:
-        [
-            user_salt,
-            checksum_salt,
-            pbkdf2_rounds,
-            user_iv,
-            master_key_blob,
-            encrypted_data,
-        ] = encrypted_backup.split(b"\n", 5)
+    [
+        user_salt,
+        checksum_salt,
+        pbkdf2_rounds,
+        user_iv,
+        master_key_blob,
+        encrypted_data,
+    ] = encrypted_backup.split(b"\n", 5)
 
-        user_salt = bytes.fromhex(user_salt.decode("utf-8"))
-        checksum_salt = bytes.fromhex(checksum_salt.decode("utf-8"))
-        pbkdf2_rounds = int(pbkdf2_rounds)
-        user_iv = bytes.fromhex(user_iv.decode("utf-8"))
-        master_key_blob = bytes.fromhex(master_key_blob.decode("utf-8"))
-    except (UnicodeDecodeError, ValueError) as exc:
-        raise AndroidBackupParsingError("Invalid encrypted backup header") from exc
+    user_salt = bytes.fromhex(user_salt.decode("utf-8"))
+    checksum_salt = bytes.fromhex(checksum_salt.decode("utf-8"))
+    pbkdf2_rounds = int(pbkdf2_rounds)
+    user_iv = bytes.fromhex(user_iv.decode("utf-8"))
+    master_key_blob = bytes.fromhex(master_key_blob.decode("utf-8"))
 
     # Derive decryption master key from password.
     master_key, master_iv = decrypt_master_key(
@@ -177,12 +174,10 @@ def parse_backup_file(data, password=None):
     if not data.startswith(b"ANDROID BACKUP"):
         raise AndroidBackupParsingError("Invalid file header")
 
-    try:
-        [_, version, is_compressed, encryption_algo, tar_data] = data.split(b"\n", 4)
-        version = int(version)
-        is_compressed = int(is_compressed)
-    except ValueError as exc:
-        raise AndroidBackupParsingError("Invalid file header") from exc
+    [_, version, is_compressed, encryption_algo, tar_data] = data.split(b"\n", 4)
+
+    version = int(version)
+    is_compressed = int(is_compressed)
 
     if encryption_algo != b"none":
         tar_data = decrypt_backup_data(

src/mvt/common/cli_plugins.py

@@ -1,297 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2026 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-
-import hashlib
-import importlib.metadata
-import importlib.util
-import logging
-import os
-import sys
-from pathlib import Path
-from types import ModuleType
-from typing import Iterable
-
-import click
-
-IOS_CLI_PLUGIN_GROUP = "mvt.ios.cli_plugins"
-ANDROID_CLI_PLUGIN_GROUP = "mvt.android.cli_plugins"
-MVT_IOS_CUSTOM_COMMANDS_ENV = "MVT_IOS_CUSTOM_COMMANDS"
-MVT_ANDROID_CUSTOM_COMMANDS_ENV = "MVT_ANDROID_CUSTOM_COMMANDS"
-
-log = logging.getLogger(__name__)
-
-
-class CustomCommandLoadError(Exception):
-    pass
-
-
-class BrokenPluginCommand(click.Command):
-    """A placeholder for an installed or configured command that failed to load."""
-
-    def __init__(self, name: str, source: str, exception: BaseException):
-        super().__init__(
-            name,
-            help=(
-                f"Unable to load external command from {source}.\n\n"
-                f"{type(exception).__name__}: {exception}"
-            ),
-            short_help="Warning: external command could not be loaded.",
-        )
-        self.source = source
-        self.exception = exception
-
-    def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
-        return args
-
-    def invoke(self, ctx: click.Context) -> None:
-        raise click.ClickException(
-            f"Unable to load external command '{self.name}' from {self.source}: "
-            f"{type(self.exception).__name__}: {self.exception}"
-        )
-
-
-def _module_name_for_path(path: Path) -> str:
-    digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
-    return f"_mvt_custom_command_{path.stem}_{digest}"
-
-
-def _iter_command_files(path: Path) -> Iterable[Path]:
-    if path.is_file():
-        if path.suffix != ".py":
-            raise CustomCommandLoadError(
-                f"Custom command file is not a Python file: {path}"
-            )
-        yield path
-        return
-
-    if path.is_dir():
-        for child in sorted(path.iterdir()):
-            if child.name.startswith(".") or child.name == "__init__.py":
-                continue
-            if child.is_file() and child.suffix == ".py":
-                yield child
-        return
-
-    raise CustomCommandLoadError(f"Custom command path does not exist: {path}")
-
-
-def _load_python_file(path: Path) -> ModuleType:
-    module_name = _module_name_for_path(path)
-    spec = importlib.util.spec_from_file_location(module_name, path)
-    if spec is None or spec.loader is None:
-        raise CustomCommandLoadError(f"Unable to load custom command file: {path}")
-
-    module = importlib.util.module_from_spec(spec)
-    sys.modules[module_name] = module
-    try:
-        spec.loader.exec_module(module)
-    except (Exception, SystemExit) as exc:
-        raise CustomCommandLoadError(
-            f"Unable to import custom command {path}: {exc}"
-        ) from exc
-
-    return module
-
-
-def load_cli_command_file(path: Path) -> click.Command:
-    module = _load_python_file(path)
-    command = getattr(module, "cli", None)
-    if not isinstance(command, click.Command):
-        raise CustomCommandLoadError(
-            f"Custom command {path} must export a Click command or group named 'cli'"
-        )
-    if command.name is None:
-        raise CustomCommandLoadError(f"Custom command {path} has no command name")
-    return command
-
-
-def _register_command(
-    group: click.Group,
-    command: click.Command,
-    *,
-    name: str,
-    source: str,
-    collision_is_error: bool,
-) -> bool:
-    if name in group.commands:
-        registered_sources = getattr(group, "_mvt_external_command_sources", {})
-        if registered_sources.get(name) == source:
-            return False
-        message = (
-            f"Unable to register external command '{name}' from {source}: "
-            "the command name is already registered"
-        )
-        if collision_is_error:
-            raise CustomCommandLoadError(message)
-        log.warning(message)
-        return False
-
-    group.add_command(command, name=name)
-    registered_sources = getattr(group, "_mvt_external_command_sources", {})
-    registered_sources[name] = source
-    setattr(group, "_mvt_external_command_sources", registered_sources)
-    return True
-
-
-def register_cli_commands_from_path(
-    group: click.Group,
-    path: str | Path,
-    *,
-    collision_is_error: bool = False,
-    failures_are_errors: bool = False,
-) -> list[str]:
-    resolved_path = Path(path).expanduser().resolve()
-    registered: list[str] = []
-
-    try:
-        command_files = list(_iter_command_files(resolved_path))
-    except CustomCommandLoadError as exc:
-        if failures_are_errors:
-            raise
-        name = resolved_path.stem.replace("_", "-") or "custom-command"
-        if _register_command(
-            group,
-            BrokenPluginCommand(name, str(resolved_path), exc),
-            name=name,
-            source=str(resolved_path),
-            collision_is_error=collision_is_error,
-        ):
-            registered.append(name)
-        return registered
-
-    for command_file in command_files:
-        try:
-            command = load_cli_command_file(command_file)
-        except CustomCommandLoadError as exc:
-            if failures_are_errors:
-                raise
-            name = command_file.stem.replace("_", "-")
-            command = BrokenPluginCommand(name, str(command_file), exc)
-
-        command_name = command.name
-        if command_name is None:
-            raise CustomCommandLoadError(
-                f"Custom command {command_file} has no command name"
-            )
-        if _register_command(
-            group,
-            command,
-            name=command_name,
-            source=str(command_file),
-            collision_is_error=collision_is_error,
-        ):
-            registered.append(command_name)
-
-    return registered
-
-
-def _entry_point_source(entry_point: importlib.metadata.EntryPoint) -> str:
-    try:
-        distribution = getattr(entry_point, "dist", None)
-        if distribution is None:
-            return f"entry point {entry_point.value}"
-
-        name = distribution.metadata.get("Name", "unknown distribution")
-        version = getattr(distribution, "version", None)
-        if version:
-            return f"{name} {version} ({entry_point.value})"
-        return f"{name} ({entry_point.value})"
-    except Exception:
-        return f"entry point {entry_point.value}"
-
-
-def register_installed_cli_commands(
-    group: click.Group,
-    entry_point_group: str,
-) -> list[str]:
-    try:
-        entry_points = importlib.metadata.entry_points(group=entry_point_group)
-    except Exception as exc:
-        log.warning(
-            "Unable to discover external commands in entry-point group %s: %s",
-            entry_point_group,
-            exc,
-        )
-        return []
-    ordered_entry_points = sorted(
-        entry_points,
-        key=lambda entry_point: (
-            entry_point.name,
-            _entry_point_source(entry_point),
-            entry_point.value,
-        ),
-    )
-    registered: list[str] = []
-
-    for entry_point in ordered_entry_points:
-        source = _entry_point_source(entry_point)
-        try:
-            command = entry_point.load()
-            if not isinstance(command, click.Command):
-                raise TypeError(
-                    f"entry point must resolve to a Click command or group, "
-                    f"not {type(command).__name__}"
-                )
-        except (Exception, SystemExit) as exc:
-            command = BrokenPluginCommand(entry_point.name, source, exc)
-
-        if _register_command(
-            group,
-            command,
-            name=entry_point.name,
-            source=source,
-            collision_is_error=False,
-        ):
-            registered.append(entry_point.name)
-
-    return registered
-
-
-def register_cli_plugins(
-    group: click.Group,
-    *,
-    entry_point_group: str,
-    environment_variable: str,
-) -> None:
-    environment_path = os.environ.get(environment_variable)
-    if environment_path:
-        register_cli_commands_from_path(group, environment_path)
-    register_installed_cli_commands(group, entry_point_group)
-
-
-def _load_command_option_callback(
-    ctx: click.Context,
-    param: click.Parameter,
-    paths: tuple[str, ...],
-) -> tuple[str, ...]:
-    if not isinstance(ctx.command, click.Group):
-        raise click.ClickException("--load-command requires a Click command group")
-
-    for path in paths:
-        try:
-            register_cli_commands_from_path(
-                ctx.command,
-                path,
-                collision_is_error=True,
-                failures_are_errors=True,
-            )
-        except CustomCommandLoadError as exc:
-            raise click.BadParameter(str(exc), ctx=ctx, param=param) from exc
-    return paths
-
-
-load_cli_commands_option = click.option(
-    "--load-command",
-    "load_commands",
-    type=click.Path(exists=True),
-    multiple=True,
-    expose_value=False,
-    is_eager=True,
-    callback=_load_command_option_callback,
-    help=(
-        "Load a custom CLI command from a Python file or folder "
-        "(can be invoked multiple times)"
-    ),
-)

src/mvt/common/command.py

@@ -20,7 +20,7 @@ from .config import settings
 from .indicators import Indicators
 from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
 from .module_loader import module_supports_command
-from .module_types import ModuleTimeline, URLResult
+from .module_types import ModuleTimeline
 from .utils import (
     CustomJSONEncoder,
     convert_datetime_to_iso,
@@ -73,7 +73,6 @@ class Command:
         self.hashes = hashes
         self.hash_values: list[dict[str, Any]] = []
         self.timeline: ModuleTimeline = []
-        self.url_results: list[URLResult] = []
 
         # Load IOCs
         self._create_storage()
@@ -151,14 +150,6 @@ class Command:
         with open(alerts_path, "w+", encoding="utf-8") as handle:
             json.dump(alerts, handle, indent=4, cls=CustomJSONEncoder)
 
-    def _store_urls(self) -> None:
-        if not self.results_path or not self.url_results:
-            return
-
-        urls_path = os.path.join(self.results_path, "urls.json")
-        with open(urls_path, "w", encoding="utf-8") as handle:
-            json.dump(self.url_results, handle, indent=4, cls=CustomJSONEncoder)
-
     def _store_alerts_timeline(self) -> None:
         if not self.results_path:
             return
@@ -405,7 +396,6 @@ class Command:
             self.executed.append(m)
             executed_by_type[module] = m
             self.timeline.extend(m.timeline)
-            self.url_results.extend(m.url_results)
             self.alertstore.extend(m.alertstore.alerts)
 
         try:
@@ -420,5 +410,4 @@ class Command:
         self._store_timeline()
         self._store_alerts_timeline()
         self._store_alerts()
-        self._store_urls()
         self._store_info()

src/mvt/common/help.py

@@ -38,7 +38,6 @@ HELP_MSG_BACKUP_KEYFILE = (
 HELP_MSG_EXTRACT_KEY = "Extract decryption key from an iTunes backup"
 HELP_MSG_CHECK_IOS_BACKUP = "Extract artifacts from an iTunes backup"
 HELP_MSG_CHECK_FS = "Extract artifacts from a full filesystem dump"
-HELP_MSG_CHECK_SYSDIAGNOSE = "Extract artifacts from an iOS sysdiagnose archive"
 
 # Android Specific
 HELP_MSG_ANDROID_BACKUP_PASSWORD = "The backup password to use for an Android backup"

src/mvt/common/indicators.py

@@ -7,10 +7,9 @@ import glob
 import json
 import logging
 import os
-from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass
 from functools import lru_cache
-from typing import Any, Dict, Iterator, List, Optional, Sequence
+from typing import Any, Dict, Iterator, List, Optional
 
 import ahocorasick
 from appdirs import user_data_dir
@@ -23,8 +22,6 @@ MVT_INDICATORS_FOLDER = os.path.join(MVT_DATA_FOLDER, "indicators")
 
 logger = logging.getLogger(__name__)
 
-URL_CHECK_MAX_WORKERS = 20
-
 
 @dataclass
 class Indicator:
@@ -49,7 +46,6 @@ class Indicators:
         self.log = log
         self.ioc_collections: List[Dict[str, Any]] = []
         self.total_ioc_count = 0
-        self.resolved_urls: Dict[str, str] = {}
 
     def _load_downloaded_indicators(self) -> None:
         if not os.path.isdir(MVT_INDICATORS_FOLDER):
@@ -440,14 +436,9 @@ class Indicators:
                         orig_url.url,
                         dest_url.url,
                     )
-                    match = self.check_url(dest_url.url)
-                    self.resolved_urls[url] = self.resolved_urls.get(
-                        dest_url.url, dest_url.url
-                    )
-                    return match
+                    return self.check_url(dest_url.url)
 
                 final_url = dest_url
-                self.resolved_urls[url] = final_url.url
             else:
                 # If it's not shortened, we just use the original URL object.
                 final_url = orig_url
@@ -488,13 +479,6 @@ class Indicators:
 
         return None
 
-    def get_expanded_url(self, url: str) -> Optional[str]:
-        """Return the final URL recorded while checking a shortened URL."""
-        expanded_url = self.resolved_urls.get(url)
-        if expanded_url and expanded_url != url:
-            return expanded_url
-        return None
-
     def check_urls(self, urls: list) -> Optional[IndicatorMatch]:
         """Check a list of URLs against the provided list of domain indicators.
 
@@ -506,41 +490,12 @@ class Indicators:
         if not urls:
             return None
 
-        return self.check_url_batches([urls])[0]
+        for url in urls:
+            check = self.check_url(url)
+            if check:
+                return check
 
-    def check_url_batches(
-        self, url_batches: Sequence[Optional[Sequence[str]]]
-    ) -> List[Optional[IndicatorMatch]]:
-        """Check batches of URLs concurrently while preserving batch order.
-
-        URLs are deduplicated across batches before checking. Each returned item
-        is the first indicator match from the corresponding input batch, using
-        the original URL order.
-        """
-        batches = [list(urls) if urls else [] for urls in url_batches]
-        unique_urls = list(
-            dict.fromkeys(url for urls in batches for url in urls)
-        )
-
-        if not unique_urls:
-            return [None] * len(batches)
-
-        if settings.NETWORK_ACCESS_ALLOWED and len(unique_urls) > 1:
-            worker_count = min(URL_CHECK_MAX_WORKERS, len(unique_urls))
-            with ThreadPoolExecutor(max_workers=worker_count) as executor:
-                url_matches = dict(
-                    zip(unique_urls, executor.map(self.check_url, unique_urls))
-                )
-        else:
-            url_matches = {url: self.check_url(url) for url in unique_urls}
-
-        return [
-            next(
-                (url_matches[url] for url in urls if url_matches[url] is not None),
-                None,
-            )
-            for urls in batches
-        ]
+        return None
 
     def check_process(self, process: str) -> Optional[IndicatorMatch]:
         """Check the provided process name against the list of process

src/mvt/common/module.py

@@ -18,7 +18,6 @@ from .module_types import (
     ModuleResults,
     ModuleSerializedResult,
     ModuleTimeline,
-    URLResult,
 )
 from .utils import CustomJSONEncoder, exec_or_profile
 
@@ -83,7 +82,6 @@ class MVTModule:
 
         self.results: ModuleResults = results if results is not None else []
         self.timeline: ModuleTimeline = []
-        self.url_results: list[URLResult] = []
         self.dependency_modules: Dict[type["MVTModule"], "MVTModule"] = {}
 
     def get_dependency_results(
@@ -112,23 +110,6 @@ class MVTModule:
     def check_indicators(self) -> None:
         raise NotImplementedError
 
-    def collect_url_results(self) -> None:
-        """Collect URL records exposed by this module."""
-
-    def add_url_result(self, url: str, timestamp: Optional[str], source: str) -> None:
-        expanded_url = None
-        if self.indicators:
-            expanded_url = self.indicators.get_expanded_url(url)
-
-        self.url_results.append(
-            {
-                "url": url,
-                "expanded_url": expanded_url,
-                "timestamp": timestamp,
-                "source": source,
-            }
-        )
-
     def save_to_json(self) -> None:
         if not self.results_path:
             return
@@ -268,15 +249,6 @@ def run_module(module: MVTModule) -> None:
                     "The %s module produced no detections!", module.__class__.__name__
                 )
 
-        try:
-            module.collect_url_results()
-        except Exception as exc:
-            module.log.exception(
-                "Error when collecting URLs from module %s: %s",
-                module.__class__.__name__,
-                exc,
-            )
-
         try:
             module.to_timeline()
         except NotImplementedError:

src/mvt/common/module_loader.py

@@ -6,7 +6,6 @@
 import hashlib
 import importlib.util
 import inspect
-import logging
 import os
 import sys
 from pathlib import Path
@@ -16,7 +15,6 @@ from typing import Iterable, Optional
 from .module import MVTModule
 
 MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES"
-log = logging.getLogger(__name__)
 
 
 class CustomModuleLoadError(Exception):
@@ -124,11 +122,6 @@ def module_supports_command(
 ) -> bool:
     supported_commands = getattr(module_class, "supported_commands", None)
     if not supported_commands:
-        log.warning(
-            "Custom module %s has no supported_commands and will not be run. "
-            "Declare the platform/command pairs it supports.",
-            module_class.__name__,
-        )
-        return False
+        return True
 
     return (platform, command) in {tuple(entry) for entry in supported_commands}

src/mvt/common/module_types.py

@@ -4,7 +4,7 @@
 #   https://license.mvt.re/1.1/
 
 from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, TypedDict, Union
+from typing import Any, Dict, List, Union
 
 
 # ModuleAtomicResult is a flexible dictionary that can contain any data.
@@ -22,13 +22,6 @@ ModuleAtomicResult = Dict[str, Any]
 ModuleResults = Any
 
 
-class URLResult(TypedDict):
-    url: str
-    expanded_url: Optional[str]
-    timestamp: Optional[str]
-    source: str
-
-
 @dataclass
 class ModuleAtomicTimeline:
     timestamp: str

src/mvt/common/url.py

@@ -5,7 +5,6 @@
 
 import logging
 from typing import Optional
-from urllib.parse import urlparse
 
 import requests
 from tld import get_tld
@@ -374,10 +373,6 @@ class URL:
         :rtype: bool
 
         """
-        parsed_url = urlparse(self.url if "://" in self.url else f"//{self.url}")
-        if self.domain.lower() == "goo.gl" and parsed_url.path.startswith("/maps/"):
-            return False
-
         if self.domain.lower() in SHORTENER_DOMAINS:
             self.is_shortened = True
 

src/mvt/common/version.py

@@ -3,4 +3,4 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-MVT_VERSION = "2026.7.29"
+MVT_VERSION = "2026.5.12"

src/mvt/ios/cli.py

@@ -8,12 +8,6 @@ import logging
 import os
 
 import click
-from mvt.common.cli_plugins import (
-    IOS_CLI_PLUGIN_GROUP,
-    MVT_IOS_CUSTOM_COMMANDS_ENV,
-    load_cli_commands_option,
-    register_cli_plugins,
-)
 from mvt.common.cmd_check_iocs import CmdCheckIOCS
 from mvt.common.completion import (
     SUPPORTED_SHELLS,
@@ -48,7 +42,6 @@ from mvt.common.help import (
     HELP_MSG_CHECK_IOCS,
     HELP_MSG_STIX2,
     HELP_MSG_CHECK_IOS_BACKUP,
-    HELP_MSG_CHECK_SYSDIAGNOSE,
     HELP_MSG_DISABLE_UPDATE_CHECK,
     HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
     HELP_MSG_COMPLETION,
@@ -57,7 +50,6 @@ from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules
 from mvt.common.password import prompt_password
 from .cmd_check_backup import CmdIOSCheckBackup
 from .cmd_check_fs import CmdIOSCheckFS
-from .cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
 from .decrypt import DecryptBackup
 from .modules.backup import BACKUP_MODULES
 from .modules.fs import FS_MODULES
@@ -92,7 +84,6 @@ def _load_custom_modules(load_module):
 # Main
 # ==============================================================================
 @click.group(invoke_without_command=False)
-@load_cli_commands_option
 @click.option(
     "--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
 )
@@ -407,77 +398,6 @@ def check_fs(
     cmd.show_support_message()
 
 
-# ==============================================================================
-# Command: check-sysdiagnose
-# ==============================================================================
-@cli.command(
-    "check-sysdiagnose",
-    context_settings=CONTEXT_SETTINGS,
-    help=HELP_MSG_CHECK_SYSDIAGNOSE,
-)
-@click.option(
-    "--iocs",
-    "-i",
-    type=click.Path(exists=True),
-    multiple=True,
-    default=[],
-    help=HELP_MSG_IOC,
-)
-@click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
-@click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
-@click.option("--module", "-m", help=HELP_MSG_MODULE)
-@click.option(
-    "--load-module",
-    type=click.Path(exists=True),
-    multiple=True,
-    default=[],
-    help=HELP_MSG_LOAD_MODULE,
-)
-@click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES)
-@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
-@click.argument("SYSDIAGNOSE_PATH", type=click.Path(exists=True))
-@click.pass_context
-def check_sysdiagnose(
-    ctx,
-    iocs,
-    output,
-    list_modules,
-    module,
-    load_module,
-    hashes,
-    verbose,
-    sysdiagnose_path,
-):
-    set_verbose_logging(verbose)
-    custom_modules = _load_custom_modules(load_module)
-    cmd = CmdIOSCheckSysdiagnose(
-        target_path=sysdiagnose_path,
-        results_path=output,
-        ioc_files=iocs,
-        module_name=module,
-        hashes=hashes,
-        disable_version_check=_get_disable_flags(ctx)[0],
-        disable_indicator_check=_get_disable_flags(ctx)[1],
-        custom_modules=custom_modules,
-    )
-
-    if not cmd._available_modules():
-        raise click.ClickException(
-            "No custom modules support mvt-ios check-sysdiagnose. "
-            "Load a module that declares supported_commands = "
-            "((\"ios\", \"check-sysdiagnose\"),)."
-        )
-
-    if list_modules:
-        cmd.list_modules()
-        return
-
-    log.info("Checking iOS sysdiagnose at path: %s", sysdiagnose_path)
-    cmd.run()
-    cmd.show_alerts_brief()
-    cmd.show_support_message()
-
-
 # ==============================================================================
 # Command: check-iocs
 # ==============================================================================
@@ -530,10 +450,3 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
 def download_iocs():
     ioc_updates = IndicatorsUpdates()
     ioc_updates.update()
-
-
-register_cli_plugins(
-    cli,
-    entry_point_group=IOS_CLI_PLUGIN_GROUP,
-    environment_variable=MVT_IOS_CUSTOM_COMMANDS_ENV,
-)

src/mvt/ios/cmd_check_sysdiagnose.py

@@ -1,173 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2023 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-
-import json
-import logging
-import os
-import shutil
-import tarfile
-from pathlib import Path, PurePosixPath
-from tempfile import TemporaryDirectory
-from typing import Any, Optional
-
-from mvt.common.command import Command
-from mvt.common.indicators import Indicators
-from mvt.common.module import MVTModule
-
-log = logging.getLogger(__name__)
-
-
-class CmdIOSCheckSysdiagnose(Command):
-    def __init__(
-        self,
-        target_path: Optional[str] = None,
-        results_path: Optional[str] = None,
-        ioc_files: Optional[list] = None,
-        iocs: Optional[Indicators] = None,
-        module_name: Optional[str] = None,
-        serial: Optional[str] = None,
-        module_options: Optional[dict] = None,
-        hashes: bool = False,
-        sub_command: bool = False,
-        disable_version_check: bool = False,
-        disable_indicator_check: bool = False,
-        custom_modules: Optional[list[type[MVTModule]]] = None,
-    ) -> None:
-        super().__init__(
-            target_path=target_path,
-            results_path=results_path,
-            ioc_files=ioc_files,
-            iocs=iocs,
-            module_name=module_name,
-            serial=serial,
-            module_options=module_options,
-            hashes=hashes,
-            sub_command=sub_command,
-            log=log,
-            disable_version_check=disable_version_check,
-            disable_indicator_check=disable_indicator_check,
-            custom_modules=custom_modules,
-        )
-        self.platform = "ios"
-        self.name = "check-sysdiagnose"
-        self.sysdiagnose_format: Optional[str] = None
-        self.sysdiagnose_archive: Optional[tarfile.TarFile] = None
-        self.sysdiagnose_files: list[str] = []
-        self.ips_files: list[dict[str, Any]] = []
-        self.temp_sysdiagnose_dir: Optional[TemporaryDirectory[str]] = None
-        self.extracted_sysdiagnose_path: Optional[str] = None
-
-    @staticmethod
-    def _parse_bugtype_header(data: bytes) -> Optional[int]:
-        try:
-            header = json.loads(data.split(b"\n", 1)[0].decode("utf-8"))
-            return int(header["bug_type"])
-        except (json.JSONDecodeError, KeyError, UnicodeDecodeError, ValueError):
-            return None
-
-    def _add_ips_file(self, file_path: str, data: bytes) -> None:
-        bug_type = self._parse_bugtype_header(data)
-        if bug_type is not None:
-            self.ips_files.append({"file_path": file_path, "bug_type": bug_type})
-
-    def init(self) -> None:
-        if not self.target_path:
-            raise ValueError("A sysdiagnose path is required")
-
-        if os.path.isdir(self.target_path):
-            self.sysdiagnose_format = "dir"
-            parent_path = Path(self.target_path).absolute().parent
-            for root, _, filenames in os.walk(self.target_path):
-                for filename in filenames:
-                    absolute_path = os.path.join(root, filename)
-                    file_path = os.path.relpath(absolute_path, parent_path)
-                    self.sysdiagnose_files.append(file_path)
-                    if filename.endswith(".ips"):
-                        with open(absolute_path, "rb") as handle:
-                            self._add_ips_file(absolute_path, handle.read())
-            return
-
-        if not os.path.isfile(self.target_path):
-            raise ValueError(f"Sysdiagnose path does not exist: {self.target_path}")
-
-        self.log.info("Parsing sysdiagnose archive. This might take a while...")
-        self.sysdiagnose_format = "tar"
-        self.sysdiagnose_archive = tarfile.open(self.target_path, "r:gz")
-        self._extract_sysdiagnose_archive()
-
-    def _extract_sysdiagnose_archive(self) -> None:
-        archive = self.sysdiagnose_archive
-        if archive is None:
-            raise RuntimeError("Sysdiagnose archive has not been initialized")
-
-        self.temp_sysdiagnose_dir = TemporaryDirectory()
-        extraction_root = Path(self.temp_sysdiagnose_dir.name).resolve()
-        archive_roots = set()
-
-        for member in archive:
-            member_path = PurePosixPath(member.name.replace("\\", "/"))
-            if member_path.is_absolute() or ".." in member_path.parts:
-                self.log.warning("Skipping unsafe sysdiagnose path %r", member.name)
-                continue
-
-            destination = extraction_root.joinpath(*member_path.parts).resolve()
-            if not destination.is_relative_to(extraction_root):
-                self.log.warning("Skipping unsafe sysdiagnose path %r", member.name)
-                continue
-
-            if not member_path.parts:
-                continue
-            archive_roots.add(member_path.parts[0])
-
-            if member.isdir():
-                destination.mkdir(parents=True, exist_ok=True)
-                continue
-
-            # Modules only need directories and regular files. Do not materialize
-            # links or device nodes from an untrusted sysdiagnose archive.
-            if not member.isfile():
-                self.log.warning("Skipping unsafe sysdiagnose member %r", member.name)
-                continue
-
-            normalized_name = member_path.as_posix()
-            self.sysdiagnose_files.append(normalized_name)
-
-            source = archive.extractfile(member)
-            if source is None:
-                continue
-
-            destination.parent.mkdir(parents=True, exist_ok=True)
-            with source, destination.open("wb") as output:
-                shutil.copyfileobj(source, output)
-
-            if normalized_name.endswith(".ips"):
-                self._add_ips_file(str(destination), destination.read_bytes())
-
-        if len(archive_roots) != 1:
-            raise ValueError("Sysdiagnose archive must contain one top-level directory")
-
-        self.extracted_sysdiagnose_path = str(extraction_root / archive_roots.pop())
-
-    def module_init(self, module) -> None:
-        module.ips_files = self.ips_files
-        if self.sysdiagnose_format == "tar":
-            if self.extracted_sysdiagnose_path is None:
-                raise RuntimeError("Sysdiagnose archive has not been extracted")
-            module.from_sysdiagnose_folder(
-                self.extracted_sysdiagnose_path, self.sysdiagnose_files
-            )
-            return
-        if self.sysdiagnose_format == "dir" and self.target_path:
-            module.from_sysdiagnose_folder(self.target_path, self.sysdiagnose_files)
-            return
-        raise RuntimeError("Sysdiagnose input has not been initialized")
-
-    def finish(self) -> None:
-        if self.sysdiagnose_archive is not None:
-            self.sysdiagnose_archive.close()
-            self.sysdiagnose_archive = None
-        if self.temp_sysdiagnose_dir is not None:
-            self.temp_sysdiagnose_dir.cleanup()
-            self.temp_sysdiagnose_dir = None

src/mvt/ios/data/ios_versions.json

@@ -1216,10 +1216,6 @@
         "version": "18.7.9",
         "build": "22H355"
     },
-    {
-        "version": "18.7.10",
-        "build": "22H373"
-    },
     {
         "version": "26",
         "build": "23A341"
@@ -1259,9 +1255,5 @@
     {
         "version": "26.5",
         "build": "23F77"
-    },
-    {
-        "version": "26.6.1",
-        "build": "23G82"
     }
 ]

src/mvt/ios/modules/base.py

@@ -9,7 +9,6 @@ import os
 import shutil
 import sqlite3
 import subprocess
-import tempfile
 from pathlib import Path
 from typing import Iterator, Optional, Union
 
@@ -21,20 +20,6 @@ from mvt.common.module import (
 )
 
 
-class TemporarySQLiteConnection(sqlite3.Connection):
-    """SQLite connection that owns a temporary copy of a database."""
-
-    temporary_directory: Optional[tempfile.TemporaryDirectory] = None
-
-    def close(self) -> None:
-        try:
-            super().close()
-        finally:
-            if self.temporary_directory:
-                self.temporary_directory.cleanup()
-                self.temporary_directory = None
-
-
 class IOSExtraction(MVTModule):
     """This class provides a base for all iOS filesystem/backup extraction
     modules."""
@@ -59,8 +44,6 @@ class IOSExtraction(MVTModule):
 
         self.is_backup = False
         self.is_fs_dump = False
-        self._recovered_sqlite_paths: dict[str, str] = {}
-        self._sqlite_temp_directories: list[tempfile.TemporaryDirectory] = []
 
     def _recover_sqlite_db_if_needed(
         self, file_path: str, forced: bool = False
@@ -70,12 +53,17 @@ class IOSExtraction(MVTModule):
         :param file_path: Path to the malformed database file.
 
         """
+        # SQLite's immutable mode cannot open databases with active WAL files.
         if not forced:
-            conn = self._open_sqlite_db(file_path)
+            # If the database is open, do not use immutable
+            if os.path.isfile(file_path + "-shm"):
+                conn = sqlite3.connect(file_path)
+            else:
+                conn = self._open_sqlite_db(file_path)
             cur = conn.cursor()
 
-            recover = False
             try:
+                recover = False
                 cur.execute("SELECT name FROM sqlite_master WHERE type='table';")
             except sqlite3.DatabaseError as exc:
                 if "database disk image is malformed" in str(exc):
@@ -94,54 +82,27 @@ class IOSExtraction(MVTModule):
             raise DatabaseCorruptedError(
                 "failed to recover without sqlite3 binary: please install sqlite3!"
             )
-        temporary_directory = tempfile.TemporaryDirectory(prefix="mvt_sqlite_recover_")
-        temporary_path = Path(temporary_directory.name)
-        source_path = temporary_path / "source.db"
-        recovered_path = temporary_path / "recovered.db"
-        shutil.copy2(file_path, source_path)
-        for suffix in ("-wal", "-shm"):
-            sidecar = Path(file_path + suffix)
-            if sidecar.is_file():
-                shutil.copy2(sidecar, Path(str(source_path) + suffix))
+        if '"' in file_path:
+            raise DatabaseCorruptedError(
+                f"database at path '{file_path}' is corrupted. unable to "
+                'recover because it has a quotation mark (") in its name'
+            )
+
+        bak_path = f"{file_path}.bak"
+        shutil.move(file_path, bak_path)
 
         ret = subprocess.call(
-            ["sqlite3", str(source_path), f'.clone "{recovered_path}"'],
+            ["sqlite3", bak_path, f'.clone "{file_path}"'],
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
         )
         if ret != 0:
-            temporary_directory.cleanup()
             raise DatabaseCorruptedError("failed to recover database")
 
-        self._sqlite_temp_directories.append(temporary_directory)
-        self._recovered_sqlite_paths[file_path] = str(recovered_path)
         self.log.info("Database at path %s recovered successfully!", file_path)
 
     def _open_sqlite_db(self, file_path: str) -> sqlite3.Connection:
-        database_path = self._recovered_sqlite_paths.get(file_path, file_path)
-        if not os.path.isfile(database_path + "-wal"):
-            uri = Path(database_path).resolve().as_uri() + "?mode=ro&immutable=1"
-            return sqlite3.connect(uri, uri=True)
-
-        temporary_directory = tempfile.TemporaryDirectory(prefix="mvt_sqlite_")
-        temporary_path = Path(temporary_directory.name) / Path(database_path).name
-        shutil.copy2(database_path, temporary_path)
-        for suffix in ("-wal", "-shm"):
-            sidecar = Path(database_path + suffix)
-            if sidecar.is_file():
-                shutil.copy2(sidecar, Path(str(temporary_path) + suffix))
-
-        try:
-            conn = sqlite3.connect(
-                temporary_path.resolve().as_uri() + "?mode=ro",
-                uri=True,
-                factory=TemporarySQLiteConnection,
-            )
-        except Exception:
-            temporary_directory.cleanup()
-            raise
-        conn.temporary_directory = temporary_directory
-        return conn
+        return sqlite3.connect(f"file:{file_path}?immutable=1", uri=True)
 
     def _get_backup_files_from_manifest(
         self, relative_path: Optional[str] = None, domain: Optional[str] = None
@@ -205,11 +166,7 @@ class IOSExtraction(MVTModule):
         if not self.target_path:
             return None
         file_path = os.path.join(self.target_path, file_id[0:2], file_id)
-        if (
-            not Path(file_path)
-            .resolve()
-            .is_relative_to(Path(self.target_path).resolve())
-        ):
+        if not Path(file_path).resolve().is_relative_to(Path(self.target_path).resolve()):
             return None
         if os.path.exists(file_path):
             return file_path
@@ -240,9 +197,9 @@ class IOSExtraction(MVTModule):
         :param backup_ids: Default value = None)
 
         """
-        file_path: Optional[str] = self.file_path
+        file_path: Optional[str] = None
         # First we check if the was an explicit file path specified.
-        if not file_path:
+        if not self.file_path:
             # Type narrowing: we know self.file_path is None here, work with local file_path
             # If not, we first try with backups.
             # We construct the path to the file according to the iTunes backup

src/mvt/ios/modules/fs/analytics.py

@@ -134,8 +134,6 @@ class Analytics(IOSExtraction):
                 isodate = ""
                 data = plistlib.loads(row[1])
                 data["isodate"] = isodate
-            else:
-                continue
 
             data["artifact"] = artifact
 

src/mvt/ios/modules/mixed/calls.py

@@ -73,9 +73,7 @@ class Calls(IOSExtraction):
                     "duration": row[1],
                     "location": row[2],
                     "number": (
-                        row[3].decode("utf-8")
-                        if isinstance(row[3], bytes)
-                        else row[3]
+                        row[3].decode("utf-8") if row[3] and row[3] is bytes else row[3]
                     ),
                     "provider": row[4],
                 }

src/mvt/ios/modules/mixed/safari_browserstate.py

@@ -19,17 +19,10 @@ from mvt.common.utils import convert_mactime_to_iso, keys_bytes_to_string
 
 from ..base import IOSExtraction
 
-# Safari profiles (iOS 17 and later) keep their per-profile databases under
-# Library/Safari/Profiles/<UUID>/, separate from the default profile's.
-SAFARI_BROWSER_STATE_BACKUP_RELPATHS = [
-    "Library/Safari/BrowserState.db",
-    "Library/Safari/Profiles/*/BrowserState.db",
-]
+SAFARI_BROWSER_STATE_BACKUP_RELPATH = "Library/Safari/BrowserState.db"
 SAFARI_BROWSER_STATE_ROOT_PATHS = [
     "private/var/mobile/Library/Safari/BrowserState.db",
-    "private/var/mobile/Library/Safari/Profiles/*/BrowserState.db",
     "private/var/mobile/Containers/Data/Application/*/Library/Safari/BrowserState.db",
-    "private/var/mobile/Containers/Data/Application/*/Library/Safari/Profiles/*/BrowserState.db",
 ]
 
 
@@ -181,22 +174,20 @@ class SafariBrowserState(IOSExtraction):
 
     def run(self) -> None:
         if self.is_backup:
-            for relative_path in SAFARI_BROWSER_STATE_BACKUP_RELPATHS:
-                for backup_file in self._get_backup_files_from_manifest(
-                    relative_path=relative_path
-                ):
-                    browserstate_path = self._get_backup_file_from_id(
-                        backup_file["file_id"]
-                    )
+            for backup_file in self._get_backup_files_from_manifest(
+                relative_path=SAFARI_BROWSER_STATE_BACKUP_RELPATH
+            ):
+                browserstate_path = self._get_backup_file_from_id(
+                    backup_file["file_id"]
+                )
 
-                    if not browserstate_path:
-                        continue
+                if not browserstate_path:
+                    continue
 
-                    self.log.info(
-                        "Found Safari browser state database at path: %s",
-                        browserstate_path,
-                    )
-                    self._process_browser_state_db(browserstate_path)
+                self.log.info(
+                    "Found Safari browser state database at path: %s", browserstate_path
+                )
+                self._process_browser_state_db(browserstate_path)
         elif self.is_fs_dump:
             for browserstate_path in self._get_fs_files_from_patterns(
                 SAFARI_BROWSER_STATE_ROOT_PATHS

src/mvt/ios/modules/mixed/safari_history.py

@@ -17,17 +17,10 @@ from mvt.common.utils import convert_mactime_to_datetime, convert_mactime_to_iso
 
 from ..base import IOSExtraction
 
-# Safari profiles (iOS 17 and later) each keep their own History.db under
-# Library/Safari/Profiles/<UUID>/, separate from the default profile's database.
-SAFARI_HISTORY_BACKUP_RELPATHS = [
-    "Library/Safari/History.db",
-    "Library/Safari/Profiles/*/History.db",
-]
+SAFARI_HISTORY_BACKUP_RELPATH = "Library/Safari/History.db"
 SAFARI_HISTORY_ROOT_PATHS = [
     "private/var/mobile/Library/Safari/History.db",
-    "private/var/mobile/Library/Safari/Profiles/*/History.db",
     "private/var/mobile/Containers/Data/Application/*/Library/Safari/History.db",
-    "private/var/mobile/Containers/Data/Application/*/Library/Safari/Profiles/*/History.db",
 ]
 
 
@@ -82,9 +75,6 @@ class SafariHistory(IOSExtraction):
 
             # We loop again through visits in order to find redirect record.
             for redirect in self.results:
-                if redirect["safari_history_db"] != result["safari_history_db"]:
-                    continue
-
                 if redirect["visit_id"] != result["redirect_destination"]:
                     continue
 
@@ -169,22 +159,17 @@ class SafariHistory(IOSExtraction):
 
     def run(self) -> None:
         if self.is_backup:
-            for relative_path in SAFARI_HISTORY_BACKUP_RELPATHS:
-                for history_file in self._get_backup_files_from_manifest(
-                    relative_path=relative_path
-                ):
-                    history_path = self._get_backup_file_from_id(
-                        history_file["file_id"]
-                    )
+            for history_file in self._get_backup_files_from_manifest(
+                relative_path=SAFARI_HISTORY_BACKUP_RELPATH
+            ):
+                history_path = self._get_backup_file_from_id(history_file["file_id"])
 
-                    if not history_path:
-                        continue
+                if not history_path:
+                    continue
 
-                    self.log.info(
-                        "Found Safari history database at path: %s", history_path
-                    )
+                self.log.info("Found Safari history database at path: %s", history_path)
 
-                    self._process_history_db(history_path)
+                self._process_history_db(history_path)
         elif self.is_fs_dump:
             for history_path in self._get_fs_files_from_patterns(
                 SAFARI_HISTORY_ROOT_PATHS

src/mvt/ios/modules/mixed/shortcuts.py

@@ -76,10 +76,8 @@ class Shortcuts(IOSExtraction):
         if not self.indicators:
             return
 
-        url_batches = [result["action_urls"] for result in self.results]
-        for result, ioc_match in zip(
-            self.results, self.indicators.check_url_batches(url_batches)
-        ):
+        for result in self.results:
+            ioc_match = self.indicators.check_urls(result["action_urls"])
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc

src/mvt/ios/modules/mixed/sms.py

@@ -85,27 +85,17 @@ class SMS(IOSExtraction):
         if not self.indicators:
             return
 
-        url_batches = []
         for result in self.results:
             message_links = result.get("links", [])
             # Making sure not link was ignored
             if message_links == []:
                 message_links = check_for_links(result.get("text", ""))
-            url_batches.append(message_links)
-
-        for result, ioc_match in zip(
-            self.results, self.indicators.check_url_batches(url_batches)
-        ):
+            ioc_match = self.indicators.check_urls(message_links)
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc
                 )
 
-    def collect_url_results(self) -> None:
-        for message in self.results:
-            for url in message.get("links", []):
-                self.add_url_result(url, message.get("isodate"), "sms")
-
     def run(self) -> None:
         self._find_ios_database(backup_ids=SMS_BACKUP_IDS, root_paths=SMS_ROOT_PATHS)
         self.log.info("Found SMS database at path: %s", self.file_path)

src/mvt/ios/modules/mixed/webkit_session_resource_log.py

@@ -70,8 +70,6 @@ class WebkitSessionResourceLog(IOSExtraction):
         if not self.indicators:
             return
 
-        records = []
-        url_batches = []
         for _, entries in self.results.items():
             for entry in entries:
                 source_domains = self._extract_domains(entry["redirect_source"])
@@ -96,42 +94,36 @@ class WebkitSessionResourceLog(IOSExtraction):
                     )
                 )
 
-                records.append((entry, source_domains, destination_domains))
-                url_batches.append(all_origins)
+                ioc_match = self.indicators.check_urls(all_origins)
+                if ioc_match:
+                    self.alertstore.critical(
+                        ioc_match.message, "", entry, matched_indicator=ioc_match.ioc
+                    )
 
-        for record, ioc_match in zip(
-            records, self.indicators.check_url_batches(url_batches)
-        ):
-            if ioc_match:
-                entry, source_domains, destination_domains = record
-                self.alertstore.critical(
-                    ioc_match.message, "", entry, matched_indicator=ioc_match.ioc
-                )
+                    redirect_path = ""
+                    if len(source_domains) > 0:
+                        redirect_path += "SOURCE: "
+                        for idx, item in enumerate(source_domains):
+                            source_domains[idx] = f'"{item}"'
 
-                redirect_path = ""
-                if len(source_domains) > 0:
-                    redirect_path += "SOURCE: "
-                    for idx, item in enumerate(source_domains):
-                        source_domains[idx] = f'"{item}"'
+                        redirect_path += ", ".join(source_domains)
+                        redirect_path += " -> "
 
-                    redirect_path += ", ".join(source_domains)
-                    redirect_path += " -> "
+                    redirect_path += f'ORIGIN: "{entry["origin"]}"'
 
-                redirect_path += f'ORIGIN: "{entry["origin"]}"'
+                    if len(destination_domains) > 0:
+                        redirect_path += " -> "
+                        redirect_path += "DESTINATION: "
+                        for idx, item in enumerate(destination_domains):
+                            destination_domains[idx] = f'"{item}"'
 
-                if len(destination_domains) > 0:
-                    redirect_path += " -> "
-                    redirect_path += "DESTINATION: "
-                    for idx, item in enumerate(destination_domains):
-                        destination_domains[idx] = f'"{item}"'
+                        redirect_path += ", ".join(destination_domains)
 
-                    redirect_path += ", ".join(destination_domains)
-
-                self.alertstore.high(
-                    f"Found HTTP redirect between suspicious domains: {redirect_path}",
-                    "",
-                    entry,
-                )
+                    self.alertstore.high(
+                        f"Found HTTP redirect between suspicious domains: {redirect_path}",
+                        "",
+                        entry,
+                    )
 
     def _extract_browsing_stats(self, log_path):
         items = []

src/mvt/ios/modules/mixed/whatsapp.py

@@ -61,20 +61,13 @@ class Whatsapp(IOSExtraction):
         if not self.indicators:
             return
 
-        url_batches = [result.get("links", []) for result in self.results]
-        for result, ioc_match in zip(
-            self.results, self.indicators.check_url_batches(url_batches)
-        ):
+        for result in self.results:
+            ioc_match = self.indicators.check_urls(result.get("links", []))
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc
                 )
 
-    def collect_url_results(self) -> None:
-        for message in self.results:
-            for url in message.get("links", []):
-                self.add_url_result(url, message.get("isodate"), "whatsapp")
-
     def run(self) -> None:
         self._find_ios_database(
             backup_ids=WHATSAPP_BACKUP_IDS, root_paths=WHATSAPP_ROOT_PATHS

src/mvt/ios/modules/net_base.py

@@ -42,8 +42,7 @@ class NetBase(IOSExtraction):
         )
 
     def _extract_net_data(self):
-        assert self.file_path is not None
-        conn = self._open_sqlite_db(self.file_path)
+        conn = sqlite3.connect(self.file_path)
         conn.row_factory = sqlite3.Row
         cur = conn.cursor()
         try:
@@ -238,7 +237,9 @@ class NetBase(IOSExtraction):
                             "been truncated in the database)"
                         )
 
-                    self.alertstore.medium(msg, proc["live_isodate"], proc)
+                    self.alertstore.medium(
+                        msg, proc["live_isodate"], proc
+                    )
 
             if not proc["live_proc_id"]:
                 self.log.info(

src/mvt/ios/modules/sysdiagnose/__init__.py

@@ -1,6 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2023 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-
-from .base import SysdiagnoseExtraction

src/mvt/ios/modules/sysdiagnose/base.py

@@ -1,105 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2023 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-
-import fnmatch
-import logging
-import os
-import re
-import tarfile
-from datetime import datetime, timezone
-from pathlib import Path
-from typing import Optional
-
-from mvt.common.module import MVTModule, ModuleResults
-
-
-class SysdiagnoseExtraction(MVTModule):
-    """Base class for custom modules that analyze an iOS sysdiagnose."""
-
-    def __init__(
-        self,
-        file_path: Optional[str] = None,
-        target_path: Optional[str] = None,
-        results_path: Optional[str] = None,
-        module_options: Optional[dict] = None,
-        log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[ModuleResults] = None,
-    ) -> None:
-        super().__init__(
-            file_path=file_path,
-            target_path=target_path,
-            results_path=results_path,
-            module_options=module_options,
-            log=log,
-            results=results,
-        )
-        self.parent_path: Optional[str] = None
-        self.files: list[str] = []
-        self.tar: Optional[tarfile.TarFile] = None
-        self.tar_files: list[str] = []
-        self.ips_files: list[dict[str, object]] = []
-
-    def from_sysdiagnose_folder(
-        self, target_path: str, sysdiagnose_files: list[str]
-    ) -> None:
-        self.parent_path = Path(target_path).absolute().parent.as_posix()
-        self.files = sysdiagnose_files
-
-    def from_sysdiagnose_tar(
-        self, sysdiagnose_archive: tarfile.TarFile, sysdiagnose_files: list[str]
-    ) -> None:
-        self.tar = sysdiagnose_archive
-        self.tar_files = sysdiagnose_files
-
-    def _extract_timezone(self):
-        """Determine the sysdiagnose timezone from its diagnostic log."""
-        file_paths = self._get_files_by_pattern("*/sysdiagnose.log")
-        if not file_paths:
-            self.log.info(
-                "Unable to determine the timezone in which the sysdiagnose was "
-                "generated. Assuming UTC for logs without a timezone."
-            )
-            return timezone.utc
-
-        content = self._get_file_content(file_paths[0]).decode(
-            "utf-8", errors="replace"
-        )
-        filenames = re.findall(r"sysdiagnose_\S+?\.tar\.gz", content)
-        if not filenames:
-            self.log.info(
-                "Unable to determine the timezone in which the sysdiagnose was "
-                "generated. Assuming UTC for logs without a timezone."
-            )
-            return timezone.utc
-
-        timestamp = "_".join(
-            filenames[0].removesuffix(".tar.gz").split("_")[1:3]
-        )
-        sysdiagnose_timezone = datetime.strptime(
-            timestamp, "%Y.%m.%d_%H-%M-%S%z"
-        ).tzinfo
-        self.log.info(
-            "Based on the sysdiagnose filename, assuming timezone %s for logs "
-            "without a timezone.",
-            sysdiagnose_timezone,
-        )
-        return sysdiagnose_timezone
-
-    def _get_files_by_pattern(self, pattern: str) -> list[str]:
-        file_names = self.tar_files if self.tar else self.files
-        return fnmatch.filter(file_names, pattern)
-
-    def _get_file_content(self, file_path: str) -> bytes:
-        if self.tar:
-            handle = self.tar.extractfile(self.tar.getmember(file_path))
-        else:
-            if self.parent_path is None:
-                raise RuntimeError("Sysdiagnose folder has not been initialized")
-            handle = open(os.path.join(self.parent_path, file_path), "rb")
-
-        if handle is None:
-            return b""
-        with handle:
-            return handle.read()

tests/android/test_artifact_dumpsys_adb.py

@@ -4,8 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact
-from mvt.android.modules.bugreport.dumpsys_adb_state import DumpsysADBState
-from mvt.common.alerts import AlertLevel
 
 from ..utils import get_artifact
 
@@ -32,66 +30,6 @@ class TestDumpsysADBArtifact:
         )
         assert user_key["user"] == "user@linux"
 
-    def test_parsing_adb_wifi(self):
-        da_adb = DumpsysADBArtifact()
-        file = get_artifact("android_data/dumpsys_adb_wifi.txt")
-        with open(file, "rb") as f:
-            data = f.read()
-
-        da_adb.parse(data)
-
-        assert len(da_adb.results) == 1
-        adb_data = da_adb.results[0]
-        assert "user_keys" in adb_data
-        assert len(adb_data["user_keys"]) == 1
-
-        user_key = adb_data["user_keys"][0]
-        assert (
-            user_key["fingerprint"] == "F0:A1:3D:8C:B3:F4:7B:09:9F:EE:8B:D8:38:2E:BD:C6"
-        )
-        assert user_key["user"] == "user@linux"
-
-        # The adb_wifi block following the keystore is not part of the keystore.
-        assert b"adb_wifi" not in adb_data["keystore"]
-
-    def test_parsing_multiline_terminated_by_structural_line(self):
-        dump_data = (
-            b"debugging_manager={\n"
-            b"  keystore=ABX\x00\x0bkeyStore\x00\x02\x11\n"
-            b"  connected_to_adb=true\n"
-            b"  adb_wifi={\n"
-            b"    enabled=false\n"
-            b"    tls_port=0\n"
-            b"  }\n"
-        )
-
-        parsed = DumpsysADBArtifact().indented_dump_parser(dump_data)
-
-        debugging_manager = parsed["debugging_manager"]
-        assert debugging_manager["keystore"] == [b"ABX\x00\x0bkeyStore\x00\x02\x11"]
-        assert debugging_manager["connected_to_adb"] == b"true"
-        assert debugging_manager["adb_wifi"] == {
-            "enabled": b"false",
-            "tls_port": b"0",
-        }
-
-    def test_parsing_multiline_terminated_by_closing_brace(self):
-        dump_data = (
-            b"debugging_manager={\n"
-            b"  keystore=ABX\x00\x0bkeyStore\x00\x02\x11\n"
-            b"}\n"
-            b"other={\n"
-            b"  value=true\n"
-            b"}\n"
-        )
-
-        parsed = DumpsysADBArtifact().indented_dump_parser(dump_data)
-
-        assert parsed["debugging_manager"]["keystore"] == [
-            b"ABX\x00\x0bkeyStore\x00\x02\x11"
-        ]
-        assert parsed["other"] == {"value": b"true"}
-
     def test_parsing_adb_xml(self):
         da_adb = DumpsysADBArtifact()
         file = get_artifact("android_data/dumpsys_adb_xml.txt")
@@ -116,119 +54,3 @@ class TestDumpsysADBArtifact:
         assert key_store_entry["user"] == "user@laptop"
         assert key_store_entry["fingerprint"] == expected_fingerprint
         assert key_store_entry["last_connected"] == "1628501829898"
-
-
-class TestDumpsysADBStateAlerts:
-    def test_no_androidqf_context_preserves_existing_behavior(self):
-        module = DumpsysADBState(
-            results=[
-                {
-                    "user_keys": [
-                        {
-                            "key": b"QUJDRA==",
-                            "user": "host@example",
-                            "fingerprint": "fingerprint",
-                        }
-                    ]
-                }
-            ]
-        )
-
-        module.check_indicators()
-
-        assert module.alertstore.alerts == []
-
-    def test_androidqf_trusted_keys_create_expected_alerts(self):
-        module = DumpsysADBState(
-            module_options={
-                "androidqf_acquisition": {
-                    "started": "2025-06-20T18:00:00Z",
-                    "adb_host_public_key": "QUJDRA== acquisition@host",
-                }
-            },
-            results=[
-                {
-                    "user_keys": [
-                        {
-                            "key": b"QUJDRA==",
-                            "user": "acquisition@host",
-                            "fingerprint": "acquisition-fingerprint",
-                        },
-                        {
-                            "key": b"RUZHSA==",
-                            "user": "other@host",
-                            "fingerprint": "other-fingerprint",
-                        },
-                        {
-                            "key": b"not-base64",
-                            "user": "invalid@host",
-                            "fingerprint": "",
-                        },
-                    ],
-                    "keystore": [
-                        {
-                            "key": b"QUJDRA==",
-                            "user": "acquisition@host",
-                            "fingerprint": "acquisition-fingerprint",
-                            "last_connected": "1750266000000",
-                        }
-                    ],
-                }
-            ],
-        )
-
-        module.check_indicators()
-
-        assert [alert.level for alert in module.alertstore.alerts] == [
-            AlertLevel.INFORMATIONAL,
-            AlertLevel.LOW,
-            AlertLevel.LOW,
-        ]
-        informational, different, invalid = module.alertstore.alerts
-        assert "at least one day before" in informational.message
-        assert informational.event_time == "2025-06-18 17:00:00.000000"
-        assert "different from the AndroidQF acquisition host" in different.message
-        assert "invalid trusted ADB host key" in invalid.message
-
-    def test_missing_androidqf_host_key_creates_low_alert(self):
-        trusted_key = {
-            "key": b"QUJDRA==",
-            "user": "host@example",
-            "fingerprint": "fingerprint",
-        }
-        module = DumpsysADBState(
-            module_options={"androidqf_acquisition": {}},
-            results=[{"user_keys": [trusted_key]}],
-        )
-
-        module.check_indicators()
-
-        assert len(module.alertstore.alerts) == 1
-        assert module.alertstore.alerts[0].level == AlertLevel.LOW
-        assert "does not include its host key" in module.alertstore.alerts[0].message
-
-    def test_recent_acquisition_host_key_does_not_create_alert(self):
-        module = DumpsysADBState(
-            module_options={
-                "androidqf_acquisition": {
-                    "started": "2025-06-20T18:00:00Z",
-                    "adb_host_public_key": "QUJDRA== acquisition@host",
-                }
-            },
-            results=[
-                {
-                    "keystore": [
-                        {
-                            "key": b"QUJDRA==",
-                            "user": "acquisition@host",
-                            "fingerprint": "fingerprint",
-                            "last_connected": "1750438800000",
-                        }
-                    ]
-                }
-            ],
-        )
-
-        module.check_indicators()
-
-        assert module.alertstore.alerts == []

tests/android/test_artifact_dumpsys_battery_history.py

@@ -42,22 +42,3 @@ class TestDumpsysBatteryHistoryArtifact:
         assert len(dba.alertstore.alerts) == 0
         dba.check_indicators()
         assert len(dba.alertstore.alerts) == 2
-
-    def test_parsing_absolute_timestamps(self):
-        dba = DumpsysBatteryHistoryArtifact()
-        dba.parse(
-            """07-15 20:27:39.431 (2) 100 +job=u0a123:"com.example/.ExampleJob"
-07-15 20:27:40.431 (2) 100 -job=u0a123:"com.example/.ExampleJob"
-"""
-        )
-
-        assert len(dba.results) == 2
-        assert dba.results[0] == {
-            "time_elapsed": "07-15 20:27:39.431",
-            "event": "start_job",
-            "uid": "u0a123",
-            "package_name": "com.example",
-            "service": "com.example/.ExampleJob",
-        }
-        assert dba.results[1]["event"] == "end_job"
-        assert dba.results[1]["uid"] == "u0a123"

tests/android/test_artifact_dumpsys_dbinfo.py

@@ -40,22 +40,3 @@ class TestDumpsysDBinfoArtifact:
         assert len(dbi.alertstore.alerts) == 0
         dbi.check_indicators()
         assert len(dbi.alertstore.alerts) == 5
-
-    def test_parsing_month_day_timestamp_without_pid(self):
-        dbi = DumpsysDBInfoArtifact()
-        dbi.parse(
-            """
-Connection pool for /data/user/0/com.example/databases/current.db:
-      Most recently executed operations:
-        0: [07-15 20:27:39.431] executeForCursorWindow took 1ms - succeeded, sql="SELECT 1"
-"""
-        )
-
-        assert dbi.results == [
-            {
-                "isodate": "07-15 20:27:39.431",
-                "action": "executeForCursorWindow",
-                "sql": "SELECT 1",
-                "path": "/data/user/0/com.example/databases/current.db",
-            }
-        ]

tests/android/test_artifact_dumpsys_receivers.py

@@ -31,44 +31,6 @@ class TestDumpsysReceiversArtifact:
             == "com.android.storagemanager"
         )
 
-    def test_parsing_misindented_action(self):
-        dr = DumpsysReceiversArtifact()
-        data = """\
-Receiver Resolver Table:
-  Non-Data Actions:
-      android.app.action.ENTER_CAR_MODE:
-        b5b40f6 com.google.android.projection.gearhead/.CarModeBroadcastReceiver
-               android.intent.action.MY_PACKAGE_REPLACED:
-        e7706c1 com.psycatgames.nhiegame/.ScheduledNotificationBootReceiver
-"""
-
-        dr.parse(data)
-
-        assert (
-            dr.results["android.intent.action.MY_PACKAGE_REPLACED"][0][
-                "package_name"
-            ]
-            == "com.psycatgames.nhiegame"
-        )
-
-    def test_parsing_misindented_first_action(self):
-        dr = DumpsysReceiversArtifact()
-        data = """\
-Receiver Resolver Table:
-  Non-Data Actions:
-               android.intent.action.MY_PACKAGE_REPLACED:
-        e7706c1 com.psycatgames.nhiegame/.ScheduledNotificationBootReceiver
-"""
-
-        dr.parse(data)
-
-        assert (
-            dr.results["android.intent.action.MY_PACKAGE_REPLACED"][0][
-                "package_name"
-            ]
-            == "com.psycatgames.nhiegame"
-        )
-
     def test_ioc_check(self, indicator_file):
         dr = DumpsysReceiversArtifact()
         file = get_artifact("android_data/dumpsys_packages.txt")

tests/android/test_artifact_tombstones.py

@@ -8,7 +8,6 @@ import datetime
 import pytest
 
 from mvt.android.artifacts.tombstone_crashes import TombstoneCrashArtifact
-from mvt.android.parsers.proto.tombstone import Tombstone
 
 from ..utils import get_artifact
 
@@ -43,73 +42,6 @@ class TestTombstoneCrashArtifact:
         assert len(tombstone_artifact.results) == 1
         self.validate_tombstone_result(tombstone_artifact.results[0])
 
-    def test_text_tombstone_preserves_abort_message(self):
-        tombstone_artifact = TombstoneCrashArtifact()
-        artifact_path = "android_data/bugreport/FS/data/tombstones/tombstone_00"
-        file = get_artifact(artifact_path)
-        with open(file, "rb") as f:
-            data = f.read()
-
-        tombstone_artifact.parse(
-            os.path.basename(artifact_path),
-            datetime.datetime(2021, 9, 29, 17, 43, 49),
-            data,
-        )
-
-        assert tombstone_artifact.results[0]["abort_message"] == (
-            "Check failed: payload.size() <= bytes_left "
-            "(payload.size()=99, bytes_left=51) "
-        )
-
-    def test_protobuf_tombstone_preserves_abort_message_and_causes(self):
-        tombstone_artifact = TombstoneCrashArtifact()
-        artifact_path = "android_data/tombstone_process.pb"
-        file = get_artifact(artifact_path)
-        with open(file, "rb") as f:
-            tombstone = Tombstone().parse(f.read())
-
-        tombstone.abort_message = "synthetic abort reason"
-        tombstone_artifact.parse_protobuf(
-            os.path.basename(artifact_path),
-            datetime.datetime(2023, 4, 12, 12, 32, 40, 518290),
-            bytes(tombstone),
-        )
-
-        result = tombstone_artifact.results[0]
-        assert result["abort_message"] == "synthetic abort reason"
-        assert result["causes"] == [
-            {
-                "human_readable": "null pointer dereference",
-                "memory_error": None,
-            }
-        ]
-
-    def test_text_tombstone_keeps_crashing_thread(self):
-        tombstone_artifact = TombstoneCrashArtifact()
-        artifact_path = "android_data/tombstone_process.txt"
-        file = get_artifact(artifact_path)
-        with open(file, "rb") as f:
-            data = f.read()
-
-        data += (
-            b"\npid: 25541, tid: 31896, name: worker-thread"
-            b"  >>> /vendor/bin/other <<<\n"
-        )
-        tombstone_artifact.parse(
-            os.path.basename(artifact_path),
-            datetime.datetime(2023, 4, 12, 12, 32, 40, 518290),
-            data,
-        )
-
-        result = tombstone_artifact.results[0]
-        assert result["pid"] == 25541
-        assert result["tid"] == 21307
-        assert result["process_name"] == "mtk.ape.decoder"
-        assert (
-            result["binary_path"]
-            == "/vendor/bin/hw/android.hardware.media.c2@1.2-mediatek"
-        )
-
     @pytest.mark.skip(reason="Not implemented yet")
     def test_tombtone_kernel_parsing(self):
         tombstone_artifact = TombstoneCrashArtifact()

tests/android/test_backup_parser.py

@@ -5,10 +5,7 @@
 
 import hashlib
 
-import pytest
-
 from mvt.android.parsers.backup import (
-    AndroidBackupParsingError,
     parse_ab_header,
     parse_backup_file,
     parse_tar_for_sms,
@@ -26,14 +23,6 @@ class TestBackupParsing:
             "encryption": None,
         }
 
-    def test_parse_truncated_encrypted_header(self):
-        with pytest.raises(
-            AndroidBackupParsingError, match="Invalid encrypted backup header"
-        ):
-            parse_backup_file(
-                b"ANDROID BACKUP\n5\n0\nAES-256\ntruncated", password="password"
-            )
-
     def test_parsing_noencryption(self):
         file = get_artifact("android_backup/backup.ab")
         with open(file, "rb") as f:

tests/common/test_cli_plugins.py

@@ -1,371 +0,0 @@
-from types import SimpleNamespace
-
-import click
-from click.testing import CliRunner
-
-from mvt.common.cli_plugins import (
-    ANDROID_CLI_PLUGIN_GROUP,
-    IOS_CLI_PLUGIN_GROUP,
-    BrokenPluginCommand,
-    load_cli_commands_option,
-    register_cli_commands_from_path,
-    register_cli_plugins,
-    register_installed_cli_commands,
-)
-
-
-COMMAND_TEMPLATE = """
-import click
-
-
-@click.command({name!r})
-@click.pass_context
-def cli(ctx):
-    click.echo({message!r})
-    if ctx.obj:
-        click.echo(ctx.obj.get("marker", ""))
-"""
-
-
-def _write_command(path, name, message="command ran"):
-    path.write_text(
-        COMMAND_TEMPLATE.format(name=name, message=message),
-        encoding="utf-8",
-    )
-    return path
-
-
-def _make_group():
-    @click.group()
-    @load_cli_commands_option
-    @click.pass_context
-    def group(ctx):
-        ctx.ensure_object(dict)
-        ctx.obj["marker"] = "parent context"
-
-    return group
-
-
-def _entry_point(name, value, command=None, exception=None, distribution="plugin"):
-    def load():
-        if exception is not None:
-            raise exception
-        return command
-
-    dist = SimpleNamespace(metadata={"Name": distribution}, version="1.0")
-    return SimpleNamespace(name=name, value=value, load=load, dist=dist)
-
-
-def test_load_command_option_registers_command_before_resolution(tmp_path):
-    command_path = _write_command(tmp_path / "hello.py", "hello")
-    group = _make_group()
-
-    result = CliRunner().invoke(
-        group,
-        ["--load-command", str(command_path), "hello"],
-    )
-
-    assert result.exit_code == 0
-    assert "command ran" in result.output
-    assert "parent context" in result.output
-
-
-def test_load_command_option_supports_folders_and_repeated_paths(tmp_path):
-    folder = tmp_path / "commands"
-    folder.mkdir()
-    _write_command(folder / "b.py", "second")
-    _write_command(folder / "a.py", "first")
-    _write_command(folder / ".hidden.py", "hidden")
-    _write_command(folder / "__init__.py", "init")
-    other = _write_command(tmp_path / "third.py", "third")
-    group = _make_group()
-
-    result = CliRunner().invoke(
-        group,
-        [
-            "--load-command",
-            str(folder),
-            "--load-command",
-            str(other),
-            "--load-command",
-            str(other),
-            "--help",
-        ],
-    )
-
-    assert result.exit_code == 0
-    assert "first" in result.output
-    assert "second" in result.output
-    assert "third" in result.output
-    assert "hidden" not in result.output
-    assert "init" not in result.output
-
-
-def test_loaded_command_participates_in_shell_completion(tmp_path):
-    command_path = _write_command(tmp_path / "hello.py", "hello")
-    group = _make_group()
-    words = f"group --load-command {command_path} he"
-
-    result = CliRunner().invoke(
-        group,
-        [],
-        env={
-            "_GROUP_COMPLETE": "bash_complete",
-            "COMP_WORDS": words,
-            "COMP_CWORD": "3",
-        },
-    )
-
-    assert result.exit_code == 0
-    assert "plain,hello" in result.output
-
-
-def test_explicit_command_import_and_contract_failures_are_usage_errors(tmp_path):
-    broken_path = tmp_path / "broken.py"
-    broken_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
-    missing_cli_path = tmp_path / "missing_cli.py"
-    missing_cli_path.write_text("value = 1", encoding="utf-8")
-
-    broken_result = CliRunner().invoke(
-        _make_group(),
-        ["--load-command", str(broken_path), "broken"],
-    )
-    missing_cli_result = CliRunner().invoke(
-        _make_group(),
-        ["--load-command", str(missing_cli_path), "missing-cli"],
-    )
-
-    assert broken_result.exit_code == 2
-    assert "broken import" in broken_result.output
-    assert missing_cli_result.exit_code == 2
-    assert "must export a Click command or group named 'cli'" in (
-        missing_cli_result.output
-    )
-
-
-def test_environment_command_failure_gets_broken_placeholder(tmp_path):
-    command_path = tmp_path / "broken_command.py"
-    command_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
-    group = click.Group()
-
-    registered = register_cli_commands_from_path(group, command_path)
-
-    assert registered == ["broken-command"]
-    assert isinstance(group.commands["broken-command"], BrokenPluginCommand)
-    result = CliRunner().invoke(group, ["broken-command"])
-    assert result.exit_code == 1
-    assert "broken import" in result.output
-    assert str(command_path) in result.output
-
-
-def test_environment_command_system_exit_gets_broken_placeholder(tmp_path):
-    command_path = tmp_path / "exiting_command.py"
-    command_path.write_text("raise SystemExit(7)", encoding="utf-8")
-    group = click.Group()
-
-    registered = register_cli_commands_from_path(group, command_path)
-
-    assert registered == ["exiting-command"]
-    assert isinstance(group.commands["exiting-command"], BrokenPluginCommand)
-    result = CliRunner().invoke(group, ["exiting-command"])
-    assert result.exit_code == 1
-    assert "Unable to import custom command" in result.output
-    assert result.output.rstrip().endswith(": 7")
-
-
-def test_installed_entry_point_name_is_the_command_name(monkeypatch):
-    @click.command("internal-name")
-    def command():
-        click.echo("installed command ran")
-
-    entry_point = _entry_point(
-        "external-name",
-        "example_plugin:cli",
-        command=command,
-    )
-    monkeypatch.setattr(
-        "mvt.common.cli_plugins.importlib.metadata.entry_points",
-        lambda **kwargs: [entry_point],
-    )
-    group = click.Group()
-
-    registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
-
-    assert registered == ["external-name"]
-    assert "internal-name" not in group.commands
-    result = CliRunner().invoke(group, ["external-name"])
-    assert result.exit_code == 0
-    assert result.output == "installed command ran\n"
-
-
-def test_broken_installed_plugin_does_not_break_cli(monkeypatch):
-    broken = _entry_point(
-        "broken",
-        "broken_plugin:cli",
-        exception=RuntimeError("missing dependency"),
-        distribution="broken-plugin",
-    )
-    monkeypatch.setattr(
-        "mvt.common.cli_plugins.importlib.metadata.entry_points",
-        lambda **kwargs: [broken],
-    )
-    group = click.Group()
-
-    register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
-
-    help_result = CliRunner().invoke(group, ["--help"])
-    assert help_result.exit_code == 0
-    assert "Warning: external command could not be loaded." in help_result.output
-
-    result = CliRunner().invoke(group, ["broken"])
-    assert result.exit_code == 1
-    assert "broken-plugin 1.0 (broken_plugin:cli)" in result.output
-    assert "RuntimeError: missing dependency" in result.output
-
-
-def test_installed_plugin_system_exit_does_not_break_cli(monkeypatch):
-    exiting = _entry_point(
-        "exiting",
-        "exiting_plugin:cli",
-        exception=SystemExit(7),
-        distribution="exiting-plugin",
-    )
-    monkeypatch.setattr(
-        "mvt.common.cli_plugins.importlib.metadata.entry_points",
-        lambda **kwargs: [exiting],
-    )
-    group = click.Group()
-
-    register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
-
-    help_result = CliRunner().invoke(group, ["--help"])
-    assert help_result.exit_code == 0
-    result = CliRunner().invoke(group, ["exiting"])
-    assert result.exit_code == 1
-    assert "SystemExit: 7" in result.output
-
-
-def test_non_click_entry_point_gets_broken_placeholder(monkeypatch):
-    invalid = _entry_point("invalid", "plugin:value", command=object())
-    monkeypatch.setattr(
-        "mvt.common.cli_plugins.importlib.metadata.entry_points",
-        lambda **kwargs: [invalid],
-    )
-    group = click.Group()
-
-    register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
-
-    assert isinstance(group.commands["invalid"], BrokenPluginCommand)
-    result = CliRunner().invoke(group, ["invalid"])
-    assert result.exit_code == 1
-    assert "must resolve to a Click command or group" in result.output
-
-
-def test_entry_point_discovery_failure_does_not_break_group(monkeypatch, caplog):
-    def fail_discovery(**kwargs):
-        raise RuntimeError("invalid package metadata")
-
-    monkeypatch.setattr(
-        "mvt.common.cli_plugins.importlib.metadata.entry_points",
-        fail_discovery,
-    )
-    group = click.Group()
-
-    registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
-
-    assert registered == []
-    assert not group.commands
-    assert "Unable to discover external commands" in caplog.text
-    assert "invalid package metadata" in caplog.text
-
-
-def test_builtin_and_first_external_command_win_collisions(monkeypatch, caplog):
-    @click.command("version")
-    def core_version():
-        pass
-
-    @click.command()
-    def first():
-        pass
-
-    @click.command()
-    def second():
-        pass
-
-    entry_points = [
-        _entry_point("duplicate", "z_plugin:cli", command=second, distribution="z"),
-        _entry_point("version", "plugin:version", command=first),
-        _entry_point("duplicate", "a_plugin:cli", command=first, distribution="a"),
-    ]
-    monkeypatch.setattr(
-        "mvt.common.cli_plugins.importlib.metadata.entry_points",
-        lambda **kwargs: entry_points,
-    )
-    group = click.Group(commands={"version": core_version})
-
-    registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
-
-    assert registered == ["duplicate"]
-    assert group.commands["version"] is core_version
-    assert group.commands["duplicate"] is first
-    assert "the command name is already registered" in caplog.text
-
-
-def test_explicit_command_cannot_replace_existing_command(tmp_path):
-    command_path = _write_command(tmp_path / "version.py", "version")
-    group = _make_group()
-
-    @group.command("version")
-    def core_version():
-        pass
-
-    result = CliRunner().invoke(
-        group,
-        ["--load-command", str(command_path), "version"],
-    )
-
-    assert result.exit_code == 2
-    assert "the command name is already registered" in result.output
-
-
-def test_platform_entry_point_groups_and_environment_paths_are_separate(
-    tmp_path, monkeypatch
-):
-    ios_path = _write_command(tmp_path / "ios.py", "ios-file")
-    android_path = _write_command(tmp_path / "android.py", "android-file")
-
-    @click.command()
-    def ios_package():
-        pass
-
-    @click.command()
-    def android_package():
-        pass
-
-    def entry_points(*, group):
-        if group == IOS_CLI_PLUGIN_GROUP:
-            return [_entry_point("ios-package", "ios_plugin:cli", ios_package)]
-        return [_entry_point("android-package", "android_plugin:cli", android_package)]
-
-    monkeypatch.setattr(
-        "mvt.common.cli_plugins.importlib.metadata.entry_points",
-        entry_points,
-    )
-    monkeypatch.setenv("TEST_IOS_COMMANDS", str(ios_path))
-    monkeypatch.setenv("TEST_ANDROID_COMMANDS", str(android_path))
-    ios_group = click.Group()
-    android_group = click.Group()
-
-    register_cli_plugins(
-        ios_group,
-        entry_point_group=IOS_CLI_PLUGIN_GROUP,
-        environment_variable="TEST_IOS_COMMANDS",
-    )
-    register_cli_plugins(
-        android_group,
-        entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
-        environment_variable="TEST_ANDROID_COMMANDS",
-    )
-
-    assert set(ios_group.commands) == {"ios-file", "ios-package"}
-    assert set(android_group.commands) == {"android-file", "android-package"}

tests/common/test_command.py

@@ -42,15 +42,6 @@ class IndependentModule(RecordingModule):
     pass
 
 
-class URLRecordingModule(RecordingModule):
-    def collect_url_results(self):
-        self.add_url_result(
-            "https://example.org/message",
-            "2026-07-29 12:00:00.000000",
-            "test-chat",
-        )
-
-
 class CustomIOSBackupModule(RecordingModule):
     supported_commands = (("ios", "check-backup"),)
 
@@ -59,10 +50,6 @@ class CustomIOSFSModule(RecordingModule):
     supported_commands = (("ios", "check-fs"),)
 
 
-class UnscopedCustomModule(RecordingModule):
-    pass
-
-
 class CustomDependsOnBuiltin(RecordingModule):
     supported_commands = (("ios", "check-backup"),)
     dependencies = (FirstModule,)
@@ -96,21 +83,6 @@ class TestCommand:
         alerts = json.loads((tmp_path / "alerts.json").read_text())
         assert alerts[0]["event"]["payload"] == "\\xa8\\xa9"
 
-    def test_stores_collected_urls(self, tmp_path):
-        cmd = RecordingCommand(results_path=str(tmp_path))
-        cmd.modules = [URLRecordingModule]
-
-        cmd.run()
-
-        assert json.loads((tmp_path / "urls.json").read_text()) == [
-            {
-                "url": "https://example.org/message",
-                "expanded_url": None,
-                "timestamp": "2026-07-29 12:00:00.000000",
-                "source": "test-chat",
-            }
-        ]
-
     def test_modules_run_in_stable_topological_order(self):
         cmd = RecordingCommand()
         cmd.modules = [ThirdModule, IndependentModule, SecondModule, FirstModule]
@@ -179,11 +151,7 @@ class TestCommand:
         cmd.platform = "ios"
         cmd.name = "check-backup"
         cmd.modules = [FirstModule]
-        cmd.custom_modules = [
-            CustomIOSBackupModule,
-            CustomIOSFSModule,
-            UnscopedCustomModule,
-        ]
+        cmd.custom_modules = [CustomIOSBackupModule, CustomIOSFSModule]
 
         assert [module.__name__ for module in cmd._ordered_modules()] == [
             "FirstModule",

tests/common/test_indicators.py

@@ -5,9 +5,7 @@
 
 import logging
 import os
-import threading
 
-import requests
 
 from mvt.common.config import settings
 from mvt.common.indicators import Indicators
@@ -82,136 +80,6 @@ class TestIndicators:
         assert ind.check_url("https://198.51.100.1:8080/")
         assert ind.check_url("https://1.1.1.1/") is None
 
-    def test_google_maps_short_url_is_not_resolved(self, indicator_file, mocker):
-        head_request = mocker.patch("mvt.common.url.requests.head")
-        ind = Indicators(log=logging)
-        ind.load_indicators_files([indicator_file], load_default=False)
-
-        assert ind.check_url("https://goo.gl/maps/example") is None
-        head_request.assert_not_called()
-
-    def test_check_url_batches_preserves_order(self, indicator_file):
-        ind = Indicators(log=logging)
-        ind.load_indicators_files([indicator_file], load_default=False)
-
-        matches = ind.check_url_batches(
-            [
-                [
-                    "https://github.com",
-                    "http://example.com/thisisbad",
-                    "https://www.example.org/foobar",
-                ],
-                ["https://github.com", "https://www.example.org/foobar"],
-                [],
-                None,
-            ]
-        )
-
-        assert matches[0]
-        assert matches[0].ioc.value == "http://example.com/thisisbad"
-        assert matches[1]
-        assert matches[1].ioc.value == "example.org"
-        assert matches[2] is None
-        assert matches[3] is None
-
-    def test_check_url_batches_deduplicates_and_limits_workers(
-        self, indicator_file, mocker
-    ):
-        ind = Indicators(log=logging)
-        ind.load_indicators_files([indicator_file], load_default=False)
-        mocker.patch("mvt.common.indicators.URL_CHECK_MAX_WORKERS", 2)
-
-        barrier = threading.Barrier(2)
-        lock = threading.Lock()
-        calls = []
-        active = 0
-        max_active = 0
-
-        def head_request(url, timeout):
-            nonlocal active, max_active
-            with lock:
-                calls.append(url)
-                active += 1
-                max_active = max(max_active, active)
-                call_number = len(calls)
-
-            try:
-                if call_number <= 2:
-                    barrier.wait(timeout=5)
-                return mocker.Mock(status_code=200, headers={})
-            finally:
-                with lock:
-                    active -= 1
-
-        mocker.patch("mvt.common.url.requests.head", side_effect=head_request)
-        urls = [
-            "https://bit.ly/one",
-            "https://tinyurl.com/two",
-            "https://t.co/three",
-        ]
-
-        assert ind.check_url_batches([urls, [urls[0]]]) == [None, None]
-        assert sorted(calls) == sorted(urls)
-        assert max_active == 2
-
-    def test_check_url_batches_respects_disabled_network(
-        self, indicator_file, mocker
-    ):
-        ind = Indicators(log=logging)
-        ind.load_indicators_files([indicator_file], load_default=False)
-        mocker.patch("mvt.common.indicators.settings.NETWORK_ACCESS_ALLOWED", False)
-        head_request = mocker.patch("mvt.common.url.requests.head")
-
-        assert ind.check_url_batches([["https://bit.ly/example"]]) == [None]
-        head_request.assert_not_called()
-
-    def test_check_url_batches_handles_nested_redirects_and_request_failures(
-        self, indicator_file, mocker
-    ):
-        ind = Indicators(log=logging)
-        ind.load_indicators_files([indicator_file], load_default=False)
-
-        def head_request(url, timeout):
-            if url == "https://bit.ly/failure":
-                raise requests.Timeout()
-            if url == "https://tinyurl.com/nested":
-                return mocker.Mock(
-                    status_code=301,
-                    headers={"Location": "https://t.co/nested"},
-                )
-            if url == "https://t.co/nested":
-                return mocker.Mock(
-                    status_code=302,
-                    headers={"Location": "https://www.example.org/landing"},
-                )
-            raise AssertionError(f"Unexpected URL: {url}")
-
-        head = mocker.patch(
-            "mvt.common.url.requests.head", side_effect=head_request
-        )
-
-        matches = ind.check_url_batches(
-            [["https://bit.ly/failure"], ["https://tinyurl.com/nested"]]
-        )
-
-        assert matches[0] is None
-        assert matches[1]
-        assert matches[1].ioc.value == "example.org"
-        assert (
-            ind.get_expanded_url("https://tinyurl.com/nested")
-            == "https://www.example.org/landing"
-        )
-        assert (
-            ind.get_expanded_url("https://t.co/nested")
-            == "https://www.example.org/landing"
-        )
-        assert ind.get_expanded_url("https://bit.ly/failure") is None
-        assert {call.args[0] for call in head.call_args_list} == {
-            "https://bit.ly/failure",
-            "https://tinyurl.com/nested",
-            "https://t.co/nested",
-        }
-
     def test_check_file_hash(self, indicator_file):
         ind = Indicators(log=logging)
         ind.load_indicators_files([indicator_file], load_default=False)

tests/common/test_module_loader.py

@@ -125,13 +125,12 @@ def test_load_custom_modules_loads_env_folder_first(tmp_path, monkeypatch):
     assert [module.__name__ for module in modules] == ["EnvModule", "CliModule"]
 
 
-def test_module_supports_command_requires_explicit_declaration(tmp_path, caplog):
+def test_module_supports_command_defaults_to_all_commands(tmp_path):
     module_path = _write_module(tmp_path / "custom.py", "DefaultModule")
     module = load_custom_modules_from_path(str(module_path))[0]
 
-    assert not module_supports_command(module, "ios", "check-backup")
-    assert not module_supports_command(module, "android", "check-bugreport")
-    assert "DefaultModule has no supported_commands" in caplog.text
+    assert module_supports_command(module, "ios", "check-backup")
+    assert module_supports_command(module, "android", "check-bugreport")
 
 
 def test_module_supports_command_honors_supported_commands(tmp_path):

tests/common/test_url.py

@@ -1,24 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2023 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-
-import pytest
-
-from mvt.common.url import URL
-
-
-@pytest.mark.parametrize(
-    "url",
-    [
-        "https://goo.gl/maps/example",
-        "http://goo.gl/maps/example?entry=message",
-        "goo.gl/maps/example",
-    ],
-)
-def test_google_maps_url_is_not_shortened(url):
-    assert URL(url).check_if_shortened() is False
-
-
-def test_other_google_short_url_is_shortened():
-    assert URL("https://goo.gl/example").check_if_shortened() is True

tests/ios_backup/test_calendar.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-import os
 
 from mvt.common.indicators import Indicators
 from mvt.common.module import run_module
@@ -22,18 +21,6 @@ class TestCalendarModule:
         assert len(m.alertstore.alerts) == 0
         assert m.results[0]["summary"] == "Super interesting meeting"
 
-    def test_calendar_with_explicit_file_path(self):
-        backup_path = get_ios_backup_folder()
-        database_path = os.path.join(
-            backup_path, "20", "2041457d5fe04d39d0ab481178355df6781e6858"
-        )
-        m = Calendar(file_path=database_path)
-
-        run_module(m)
-
-        assert len(m.results) == 1
-        assert m.results[0]["summary"] == "Super interesting meeting"
-
     def test_calendar_detection(self, indicator_file):
         m = Calendar(target_path=get_ios_backup_folder())
         ind = Indicators(log=logging.getLogger())

tests/ios_backup/test_safari_browserstate.py

@@ -4,44 +4,12 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-import shutil
-
-import pytest
 
 from mvt.common.indicators import Indicators
 from mvt.common.module import run_module
 from mvt.ios.modules.mixed.safari_browserstate import SafariBrowserState
 
-from ..utils import add_backup_manifest_entry, get_ios_backup_folder
-
-# fileID of HomeDomain::Library/Safari/BrowserState.db in the test backup.
-DEFAULT_BROWSER_STATE_FILE_ID = "3a47b0981ed7c10f3e2800aa66bac96a3b5db28e"
-PROFILE_UUID = "00000000-0000-4000-A000-000000000001"
-PROFILE_BROWSER_STATE_FILE_ID = "bb00000000000000000000000000000000000001"
-
-
-@pytest.fixture
-def backup_with_safari_profile(tmp_path):
-    """An iTunes backup where a Safari profile has its own browser state."""
-    backup_path = tmp_path / "backup"
-    shutil.copytree(get_ios_backup_folder(), backup_path)
-
-    profile_db = (
-        backup_path / PROFILE_BROWSER_STATE_FILE_ID[:2] / PROFILE_BROWSER_STATE_FILE_ID
-    )
-    profile_db.parent.mkdir(parents=True, exist_ok=True)
-    shutil.copyfile(
-        backup_path / DEFAULT_BROWSER_STATE_FILE_ID[:2] / DEFAULT_BROWSER_STATE_FILE_ID,
-        profile_db,
-    )
-    add_backup_manifest_entry(
-        backup_path,
-        PROFILE_BROWSER_STATE_FILE_ID,
-        "AppDomain-com.apple.mobilesafari",
-        f"Library/Safari/Profiles/{PROFILE_UUID}/BrowserState.db",
-    )
-
-    return str(backup_path)
+from ..utils import get_ios_backup_folder
 
 
 class TestSafariBrowserStateModule:
@@ -53,15 +21,6 @@ class TestSafariBrowserStateModule:
         assert len(m.timeline) == 1
         assert len(m.alertstore.alerts) == 0
 
-    def test_parsing_backup_with_profile(self, backup_with_safari_profile):
-        m = SafariBrowserState(target_path=backup_with_safari_profile)
-        m.is_backup = True
-        run_module(m)
-
-        # Both the default profile and the named profile are extracted.
-        assert len(m.results) == 2
-        assert len({result["safari_browser_state_db"] for result in m.results}) == 2
-
     def test_detection(self, indicator_file):
         m = SafariBrowserState(target_path=get_ios_backup_folder())
         m.is_backup = True

tests/ios_backup/test_safari_history.py

@@ -1,165 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2023 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-
-import logging
-import shutil
-import sqlite3
-from pathlib import Path
-
-import pytest
-
-from mvt.common.indicators import Indicators
-from mvt.common.module import run_module
-from mvt.ios.modules.mixed.safari_history import SafariHistory
-
-from ..utils import add_backup_manifest_entry, get_ios_backup_folder
-
-# fileID of HomeDomain::Library/Safari/History.db in the test backup.
-DEFAULT_HISTORY_FILE_ID = "1a0e7afc19d307da602ccdcece51af33afe92c53"
-PROFILE_UUID = "00000000-0000-4000-A000-000000000001"
-PROFILE_HISTORY_FILE_ID = "aa00000000000000000000000000000000000001"
-
-# example.org is already a test indicator, so use domains that do not match.
-DEFAULT_URL = "https://default.example.net/visited-page"
-PROFILE_URL = "https://profile.example.net/visited-page"
-
-
-def create_history_db(path, url):
-    """Create a minimal Safari History.db holding a single visit."""
-    path.parent.mkdir(parents=True, exist_ok=True)
-    conn = sqlite3.connect(path)
-    conn.executescript(
-        """
-        CREATE TABLE history_items (id INTEGER PRIMARY KEY, url TEXT);
-        CREATE TABLE history_visits (
-            id INTEGER PRIMARY KEY,
-            history_item INTEGER,
-            visit_time REAL,
-            redirect_source INTEGER,
-            redirect_destination INTEGER
-        );
-        """
-    )
-    conn.execute("INSERT INTO history_items VALUES (1, ?);", (url,))
-    conn.execute("INSERT INTO history_visits VALUES (1, 1, 726100000.0, NULL, NULL);")
-    conn.commit()
-    conn.close()
-
-
-@pytest.fixture
-def backup_with_safari_profile(tmp_path):
-    """An iTunes backup where Safari has both a default and a named profile."""
-    backup_path = tmp_path / "backup"
-    shutil.copytree(get_ios_backup_folder(), backup_path)
-
-    # The default profile's database ships empty, so give it a visit to make
-    # sure the profile lookup does not replace the pre-existing one.
-    create_history_db(
-        backup_path / DEFAULT_HISTORY_FILE_ID[:2] / DEFAULT_HISTORY_FILE_ID,
-        DEFAULT_URL,
-    )
-
-    create_history_db(
-        backup_path / PROFILE_HISTORY_FILE_ID[:2] / PROFILE_HISTORY_FILE_ID,
-        PROFILE_URL,
-    )
-    add_backup_manifest_entry(
-        backup_path,
-        PROFILE_HISTORY_FILE_ID,
-        "AppDomain-com.apple.mobilesafari",
-        f"Library/Safari/Profiles/{PROFILE_UUID}/History.db",
-    )
-
-    return str(backup_path)
-
-
-@pytest.fixture
-def fs_dump_with_safari_profile(tmp_path):
-    """A filesystem dump where Safari has both a default and a named profile."""
-    safari_path = tmp_path / "private" / "var" / "mobile" / "Library" / "Safari"
-    profile_path = safari_path / "Profiles" / PROFILE_UUID
-    profile_path.mkdir(parents=True)
-
-    create_history_db(safari_path / "History.db", DEFAULT_URL)
-    create_history_db(profile_path / "History.db", PROFILE_URL)
-
-    return str(tmp_path)
-
-
-class TestSafariHistoryModule:
-    def test_parsing(self):
-        m = SafariHistory(target_path=get_ios_backup_folder())
-        m.is_backup = True
-        run_module(m)
-        assert len(m.results) == 0
-        assert len(m.alertstore.alerts) == 0
-
-    def test_parsing_backup_with_profile(self, backup_with_safari_profile):
-        m = SafariHistory(target_path=backup_with_safari_profile)
-        m.is_backup = True
-        run_module(m)
-
-        # Both the default profile and the named profile are extracted.
-        assert len(m.results) == 2
-        assert {result["url"] for result in m.results} == {DEFAULT_URL, PROFILE_URL}
-        assert len({result["safari_history_db"] for result in m.results}) == 2
-
-    def test_parsing_fs_dump_with_profile(self, fs_dump_with_safari_profile):
-        m = SafariHistory(target_path=fs_dump_with_safari_profile)
-        m.is_fs_dump = True
-        run_module(m)
-
-        assert len(m.results) == 2
-        assert {result["url"] for result in m.results} == {DEFAULT_URL, PROFILE_URL}
-
-    def test_redirect_ids_are_scoped_to_database(self, fs_dump_with_safari_profile):
-        safari_path = (
-            Path(fs_dump_with_safari_profile)
-            / "private"
-            / "var"
-            / "mobile"
-            / "Library"
-            / "Safari"
-        )
-        with sqlite3.connect(safari_path / "History.db") as conn:
-            conn.execute(
-                "UPDATE history_items SET url = ? WHERE id = 1;",
-                ("http://safe.example.com/start",),
-            )
-            conn.execute(
-                "INSERT INTO history_items VALUES (2, ?);",
-                ("https://safe.example.com/end",),
-            )
-            conn.execute(
-                "UPDATE history_visits SET redirect_destination = 2 WHERE id = 1;"
-            )
-            conn.execute(
-                "INSERT INTO history_visits VALUES (2, 2, 726100000.1, 1, NULL);"
-            )
-
-        profile_db = safari_path / "Profiles" / PROFILE_UUID / "History.db"
-        with sqlite3.connect(profile_db) as conn:
-            # Visit IDs are local to each database and commonly overlap.
-            conn.execute("UPDATE history_visits SET id = 2 WHERE id = 1;")
-
-        m = SafariHistory(target_path=fs_dump_with_safari_profile)
-        m.is_fs_dump = True
-        run_module(m)
-
-        assert len(m.results) == 3
-        assert len(m.alertstore.alerts) == 0
-
-    def test_detection_in_profile(self, backup_with_safari_profile, indicator_file):
-        """An indicator only visited inside a Safari profile still alerts."""
-        m = SafariHistory(target_path=backup_with_safari_profile)
-        m.is_backup = True
-        ind = Indicators(log=logging.getLogger())
-        ind.parse_stix2(indicator_file)
-        ind.ioc_collections[0]["domains"].append("profile.example.net")
-        m.indicators = ind
-        run_module(m)
-
-        assert len(m.alertstore.alerts) == 1
-        assert m.alertstore.alerts[0].event["url"] == PROFILE_URL

tests/ios_backup/test_sms.py

@@ -18,14 +18,6 @@ class TestSMSModule:
         run_module(m)
         assert len(m.results) == 1
         assert len(m.timeline) == 2
-        assert m.url_results == [
-            {
-                "url": "https://badbadbad.example.org/",
-                "expanded_url": None,
-                "timestamp": "2019-08-29 23:13:30.000000",
-                "source": "sms",
-            }
-        ]
         assert len(m.alertstore.alerts) == 0
 
     def test_detection(self, indicator_file):
@@ -37,31 +29,3 @@ class TestSMSModule:
         m.indicators = ind
         run_module(m)
         assert len(m.alertstore.alerts) == 1
-
-    def test_detection_batches_urls_and_preserves_event(self, indicator_file, mocker):
-        results = [
-            {
-                "text": "first",
-                "links": ["http://example.com/thisisbad"],
-            },
-            {
-                "text": "second",
-                "links": ["https://github.com"],
-            },
-        ]
-        m = SMS(results=results)
-        ind = Indicators(log=logging.getLogger())
-        ind.parse_stix2(indicator_file)
-        batch_check = mocker.spy(ind, "check_url_batches")
-        m.indicators = ind
-
-        m.check_indicators()
-
-        batch_check.assert_called_once_with(
-            [
-                ["http://example.com/thisisbad"],
-                ["https://github.com"],
-            ]
-        )
-        assert len(m.alertstore.alerts) == 1
-        assert m.alertstore.alerts[0].event is results[0]

tests/ios_backup/test_sqlite_handling.py

@@ -1,85 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2023 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-
-import hashlib
-import os
-import plistlib
-import shutil
-import sqlite3
-
-from mvt.ios.modules.base import IOSExtraction
-from mvt.ios.modules.fs.analytics import Analytics
-
-
-def _sha256(path):
-    return hashlib.sha256(path.read_bytes()).hexdigest()
-
-
-def test_open_sqlite_reads_wal_without_modifying_evidence(tmp_path):
-    live_path = tmp_path / "live.db"
-    evidence_path = tmp_path / "evidence.db"
-    conn = sqlite3.connect(live_path)
-    conn.execute("PRAGMA journal_mode=WAL;")
-    conn.execute("PRAGMA wal_autocheckpoint=0;")
-    conn.execute("CREATE TABLE records (value TEXT);")
-    conn.commit()
-    conn.execute("INSERT INTO records VALUES ('from wal');")
-    conn.commit()
-
-    shutil.copy2(live_path, evidence_path)
-    shutil.copy2(str(live_path) + "-wal", str(evidence_path) + "-wal")
-    conn.close()
-
-    evidence_hash = _sha256(evidence_path)
-    wal_hash = _sha256(tmp_path / "evidence.db-wal")
-    module = IOSExtraction(file_path=str(evidence_path))
-    read_conn = module._open_sqlite_db(str(evidence_path))
-    rows = read_conn.execute("SELECT value FROM records;").fetchall()
-    read_conn.close()
-
-    assert rows == [("from wal",)]
-    assert _sha256(evidence_path) == evidence_hash
-    assert _sha256(tmp_path / "evidence.db-wal") == wal_hash
-    assert not os.path.exists(str(evidence_path) + "-shm")
-
-
-def test_recovery_preserves_source_database(tmp_path):
-    database_path = tmp_path / "source.db"
-    conn = sqlite3.connect(database_path)
-    conn.execute("CREATE TABLE records (value TEXT);")
-    conn.execute("INSERT INTO records VALUES ('preserved');")
-    conn.commit()
-    conn.close()
-    source_hash = _sha256(database_path)
-
-    module = IOSExtraction(file_path=str(database_path))
-    module._recover_sqlite_db_if_needed(str(database_path), forced=True)
-    recovered_conn = module._open_sqlite_db(str(database_path))
-    rows = recovered_conn.execute("SELECT value FROM records;").fetchall()
-    recovered_conn.close()
-
-    assert rows == [("preserved",)]
-    assert _sha256(database_path) == source_hash
-    assert not os.path.exists(str(database_path) + ".bak")
-
-
-def test_analytics_skips_empty_rows_and_continues(tmp_path):
-    database_path = tmp_path / "analytics.db"
-    conn = sqlite3.connect(database_path)
-    for table in ("hard_failures", "soft_failures", "all_events"):
-        conn.execute(f"CREATE TABLE {table} (timestamp REAL, data BLOB);")
-    conn.execute("INSERT INTO hard_failures VALUES (NULL, NULL);")
-    conn.execute(
-        "INSERT INTO soft_failures VALUES (?, ?);",
-        (1.0, plistlib.dumps({"event": "valid"})),
-    )
-    conn.commit()
-    conn.close()
-
-    module = Analytics(file_path=str(database_path))
-    module._extract_analytics_data()
-
-    assert len(module.results) == 1
-    assert module.results[0]["event"] == "valid"

tests/ios_backup/test_whatsapp.py

@@ -1,35 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2026 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-
-import logging
-
-from mvt.common.indicators import Indicators
-from mvt.ios.modules.mixed.whatsapp import Whatsapp
-
-
-def test_collect_url_results_includes_expansion():
-    module = Whatsapp(
-        results=[
-            {
-                "links": ["https://bit.ly/message"],
-                "isodate": "2026-07-29 12:00:00.000000",
-            }
-        ]
-    )
-    module.indicators = Indicators(log=logging.getLogger())
-    module.indicators.resolved_urls["https://bit.ly/message"] = (
-        "https://example.org/landing"
-    )
-
-    module.collect_url_results()
-
-    assert module.url_results == [
-        {
-            "url": "https://bit.ly/message",
-            "expanded_url": "https://example.org/landing",
-            "timestamp": "2026-07-29 12:00:00.000000",
-            "source": "whatsapp",
-        }
-    ]

tests/test_check_android_androidqf.py

@@ -3,19 +3,13 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-import json
 import logging
 import os
 import shutil
-import tempfile
-import zipfile
 
 from click.testing import CliRunner
 
 from mvt.android.cli import check_androidqf
-from mvt.android.cmd_check_androidqf import CmdAndroidCheckAndroidQF
-from mvt.android.modules.androidqf import ANDROIDQF_MODULES
-from mvt.android.modules.androidqf.aqf_log_timestamps import AQFLogTimestamps
 from mvt.common.config import settings
 
 from .utils import get_artifact_folder
@@ -24,76 +18,12 @@ TEST_BACKUP_PASSWORD = "123456"
 
 
 class TestCheckAndroidqfCommand:
-    def test_log_timestamps_module_is_registered(self):
-        assert AQFLogTimestamps in ANDROIDQF_MODULES
-
     def test_check(self):
         runner = CliRunner()
         path = os.path.join(get_artifact_folder(), "androidqf")
         result = runner.invoke(check_androidqf, [path])
         assert result.exit_code == 0
 
-    def test_check_stores_nested_sms_urls(self, tmp_path):
-        runner = CliRunner()
-        path = os.path.join(get_artifact_folder(), "androidqf")
-
-        result = runner.invoke(check_androidqf, ["--output", str(tmp_path), path])
-
-        assert result.exit_code == 0
-        urls = json.loads((tmp_path / "urls.json").read_text())
-        assert {entry["url"] for entry in urls} == {
-            "http://google.com",
-            "https://google.com/",
-        }
-        assert all(
-            set(entry) == {"url", "expanded_url", "timestamp", "source"}
-            for entry in urls
-        )
-        assert all(entry["source"] == "sms" for entry in urls)
-
-    def test_acquisition_context_is_passed_to_bugreport(self, tmp_path, mocker):
-        data_path = tmp_path / "androidqf"
-        data_path.mkdir()
-        (data_path / "acquisition.json").write_text(
-            json.dumps(
-                {
-                    "started": "2025-06-20T18:00:00Z",
-                    "adb_host_public_key": "QUJDRA== acquisition@host",
-                }
-            )
-        )
-        with zipfile.ZipFile(data_path / "bugreport.zip", "w"):
-            pass
-
-        nested_command = mocker.patch(
-            "mvt.android.cmd_check_androidqf.CmdAndroidCheckBugreport"
-        )
-        nested_command.return_value.timeline = []
-        nested_command.return_value.alertstore.alerts = []
-        command = CmdAndroidCheckAndroidQF(target_path=str(data_path))
-        command.init()
-
-        assert command.run_bugreport_cmd() is True
-
-        assert nested_command.call_args.kwargs["module_options"][
-            "androidqf_acquisition"
-        ] == {
-            "started": "2025-06-20T18:00:00Z",
-            "adb_host_public_key": "QUJDRA== acquisition@host",
-        }
-
-    def test_acquisition_context_falls_back_to_public_key_file(self, tmp_path):
-        data_path = tmp_path / "androidqf"
-        data_path.mkdir()
-        (data_path / "adb_host_key.pub").write_text("QUJDRA== acquisition@host\n")
-        command = CmdAndroidCheckAndroidQF(target_path=str(data_path))
-
-        command.init()
-
-        assert command.module_options["androidqf_acquisition"] == {
-            "adb_host_public_key": "QUJDRA== acquisition@host\n"
-        }
-
     def test_check_encrypted_backup_prompt_valid(self, mocker):
         """Prompt for password on CLI"""
         prompt_mock = mocker.patch(
@@ -159,25 +89,3 @@ class TestCheckAndroidqfCommand:
         assert not any(
             record.levelname in {"CRITICAL", "FATAL"} for record in caplog.records
         )
-
-    def test_intrusion_log_zip_rejects_path_traversal(self, tmp_path, mocker, caplog):
-        escaped_name = f"mvt-escaped-{tmp_path.name}.txt"
-        escaped_path = os.path.join(tempfile.gettempdir(), escaped_name)
-        archive_path = tmp_path / "androidqf.zip"
-        with zipfile.ZipFile(archive_path, "w") as archive:
-            archive.writestr(f"intrusion_logs/../{escaped_name}", "unsafe")
-            archive.writestr("intrusion_logs/safe.txt", "safe")
-
-        nested_command = mocker.patch(
-            "mvt.android.cmd_check_androidqf.CmdAndroidCheckIntrusionLogs"
-        )
-        nested_command.return_value.timeline = []
-        nested_command.return_value.alertstore.alerts = []
-        command = CmdAndroidCheckAndroidQF(target_path=str(archive_path))
-        command.init()
-
-        with caplog.at_level(logging.WARNING):
-            assert command.run_intrusion_logs_cmd() is True
-
-        assert not os.path.exists(escaped_path)
-        assert "Skipping unsafe intrusion log archive entry" in caplog.text

tests/test_check_android_bugreport.py

@@ -18,13 +18,3 @@ class TestCheckBugreportCommand:
         path = os.path.join(get_artifact_folder(), "android_data/bugreport/")
         result = runner.invoke(check_bugreport, [path])
         assert result.exit_code == 0
-
-    def test_invalid_zip_reports_clean_error(self, tmp_path):
-        path = tmp_path / "invalid.zip"
-        path.write_bytes(b"not a zip archive")
-
-        result = CliRunner().invoke(check_bugreport, [str(path)])
-
-        assert result.exit_code == 1
-        assert "Invalid bugreport archive" in result.output
-        assert "Traceback" not in result.output

tests/test_check_ios_sysdiagnose.py

@@ -1,57 +0,0 @@
-from click.testing import CliRunner
-
-from mvt.ios.cli import check_sysdiagnose
-
-
-CUSTOM_MODULE = """
-from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
-
-
-class CustomSysdiagnoseModule(SysdiagnoseExtraction):
-    supported_commands = (("ios", "check-sysdiagnose"),)
-    slug = "custom_sysdiagnose_module"
-
-    def run(self):
-        file_path = self._get_files_by_pattern("*/artifact.txt")[0]
-        self.results = [{"content": self._get_file_content(file_path).decode("utf-8")}]
-
-    def check_indicators(self):
-        pass
-
-    def serialize(self, result):
-        return None
-"""
-
-
-def _create_sysdiagnose_folder(tmp_path):
-    folder = tmp_path / "sysdiagnose"
-    folder.mkdir()
-    (folder / "artifact.txt").write_text("artifact", encoding="utf-8")
-    return folder
-
-
-def test_check_sysdiagnose_runs_explicitly_scoped_custom_module(tmp_path):
-    module_path = tmp_path / "custom_sysdiagnose.py"
-    module_path.write_text(CUSTOM_MODULE, encoding="utf-8")
-    output_path = tmp_path / "output"
-
-    result = CliRunner().invoke(
-        check_sysdiagnose,
-        [
-            "--load-module",
-            str(module_path),
-            "--output",
-            str(output_path),
-            str(_create_sysdiagnose_folder(tmp_path)),
-        ],
-    )
-
-    assert result.exit_code == 0
-    assert (output_path / "custom_sysdiagnose_module.json").exists()
-
-
-def test_check_sysdiagnose_requires_an_explicitly_scoped_module(tmp_path):
-    result = CliRunner().invoke(check_sysdiagnose, [str(_create_sysdiagnose_folder(tmp_path))])
-
-    assert result.exit_code != 0
-    assert "No custom modules support mvt-ios check-sysdiagnose" in result.output

tests/test_cmd_check_sysdiagnose.py

@@ -1,117 +0,0 @@
-import io
-import tarfile
-from datetime import timedelta
-from pathlib import Path
-
-from mvt.ios.cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
-from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
-
-
-class SysdiagnoseTestModule(SysdiagnoseExtraction):
-    supported_commands = (("ios", "check-sysdiagnose"),)
-
-    def run(self):
-        file_path = self._get_files_by_pattern("*/artifact.txt")[0]
-        self.results = [
-            {
-                "content": self._get_file_content(file_path).decode("utf-8"),
-                "timezone_offset": self._extract_timezone().utcoffset(None).seconds,
-            }
-        ]
-
-    def check_indicators(self):
-        pass
-
-    def serialize(self, result):
-        return None
-
-
-def _create_sysdiagnose_folder(tmp_path):
-    folder = tmp_path / "sysdiagnose"
-    folder.mkdir()
-    (folder / "artifact.txt").write_text("artifact", encoding="utf-8")
-    (folder / "sysdiagnose.log").write_text(
-        "sysdiagnose_2024.01.02_03-04-05+0200.tar.gz", encoding="utf-8"
-    )
-    (folder / "report.ips").write_text('{"bug_type": 210}\nbody', encoding="utf-8")
-    return folder
-
-
-def _create_sysdiagnose_archive(tmp_path, folder):
-    archive_path = tmp_path / "sysdiagnose.tar.gz"
-    with tarfile.open(archive_path, "w:gz") as archive:
-        for path in folder.iterdir():
-            archive.add(path, arcname=f"sysdiagnose/{path.name}")
-    return archive_path
-
-
-def _run_command(path):
-    command = CmdIOSCheckSysdiagnose(
-        target_path=str(path), custom_modules=[SysdiagnoseTestModule]
-    )
-    command.run()
-    return command
-
-
-def test_check_sysdiagnose_from_folder(tmp_path):
-    command = _run_command(_create_sysdiagnose_folder(tmp_path))
-
-    assert command.executed[0].results == [
-        {"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
-    ]
-    assert command.executed[0].ips_files == [
-        {"file_path": str(tmp_path / "sysdiagnose" / "report.ips"), "bug_type": 210}
-    ]
-
-
-def test_check_sysdiagnose_from_archive_closes_archive(tmp_path):
-    folder = _create_sysdiagnose_folder(tmp_path)
-    command = _run_command(_create_sysdiagnose_archive(tmp_path, folder))
-
-    assert command.executed[0].results == [
-        {"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
-    ]
-    assert command.executed[0].ips_files == [
-        {
-            "file_path": str(
-                Path(command.extracted_sysdiagnose_path) / "report.ips"
-            ),
-            "bug_type": 210,
-        }
-    ]
-    assert command.sysdiagnose_archive is None
-
-
-def test_archive_is_extracted_once_and_unsafe_members_are_skipped(tmp_path):
-    archive_path = tmp_path / "sysdiagnose.tar.gz"
-    escaped_path = tmp_path / "escaped.txt"
-    content = b"test content"
-    member = tarfile.TarInfo("sysdiagnose/artifact.txt")
-    member.size = len(content)
-
-    with tarfile.open(archive_path, "w:gz") as archive:
-        archive.addfile(member, io.BytesIO(content))
-        escaped = tarfile.TarInfo(f"sysdiagnose/../../{escaped_path.name}")
-        escaped.size = len(content)
-        archive.addfile(escaped, io.BytesIO(content))
-        link = tarfile.TarInfo("sysdiagnose/link")
-        link.type = tarfile.SYMTYPE
-        link.linkname = "/etc/hostname"
-        archive.addfile(link)
-
-    command = CmdIOSCheckSysdiagnose(target_path=str(archive_path))
-    try:
-        command.init()
-        extracted_path = Path(command.extracted_sysdiagnose_path)
-        assert (extracted_path / "artifact.txt").read_bytes() == content
-        assert not escaped_path.exists()
-        assert not (extracted_path / "link").exists()
-
-        module = SysdiagnoseExtraction()
-        command.module_init(module)
-        assert module.tar is None
-        assert module.parent_path == str(extracted_path.parent)
-    finally:
-        command.finish()
-
-    assert not extracted_path.exists()

tests/utils.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 import os
-import sqlite3
 from pathlib import Path
 
 
@@ -38,22 +37,6 @@ def get_indicator_file():
     print("PYTEST env", os.getenv("PYTEST_CURRENT_TEST"))
 
 
-def add_backup_manifest_entry(backup_path, file_id, domain, relative_path):
-    """
-    Register an extra file in a test backup's Manifest.db
-    """
-    conn = sqlite3.connect(os.path.join(backup_path, "Manifest.db"))
-    conn.execute(
-        "INSERT INTO Files (fileID, domain, relativePath, flags, file) "
-        "VALUES (?, ?, ?, 1, ?);",
-        (file_id, domain, relative_path, b""),
-    )
-    conn.commit()
-    # Checkpoint the test change so the fixture does not retain SQLite sidecars.
-    conn.execute("PRAGMA journal_mode=DELETE;")
-    conn.close()
-
-
 def delete_tmp_db_files(file_path):
     """
     Remove Sqlite temporary files that appear on some platforms