Document network access options

docs/development.md

@@ -21,24 +21,6 @@ make check
 
 Run these tests before making new commits or opening pull requests.
 
-## Module dependencies
-
-Modules can require other modules to run first by declaring their classes in
-`dependencies`. The command runner uses a stable topological ordering, so the
-existing module list order is preserved wherever dependency constraints allow.
-
-```python
-class DependentModule(MVTModule):
-    dependencies = (PrerequisiteModule,)
-
-    def run(self):
-        prerequisite_results = self.get_dependency_results(PrerequisiteModule)
-```
-
-Selecting a single module also runs its transitive dependencies. If a dependency
-is unavailable or the dependency graph contains a cycle, the command logs a
-warning and does not run any modules.
-
 ## Profiling
 
 Some MVT modules extract and process significant amounts of data during the analysis process or while checking results against known indicators. Care must be

src/mvt/android/modules/intrusion_logs/security_event.py

@@ -264,7 +264,6 @@ SECURITY_EVENT_TAGS = {
 }
 
 SECURITY_EVENT_METADATA_KEYS = {
-    "event_id",
     "event_time",
     "event_type",
     "timestamp",

src/mvt/common/command.py

@@ -8,7 +8,6 @@ import logging
 import os
 import sys
 from datetime import datetime
-from heapq import heappop, heappush
 from typing import Any, Optional
 
 from rich.console import Console
@@ -19,7 +18,6 @@ from .alerts import AlertLevel, AlertStore
 from .config import settings
 from .indicators import Indicators
 from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
-from .module_types import ModuleTimeline
 from .utils import (
     CustomJSONEncoder,
     convert_datetime_to_iso,
@@ -46,7 +44,7 @@ class Command:
         disable_indicator_check: bool = False,
     ) -> None:
         self.name = ""
-        self.modules: list[type[MVTModule]] = []
+        self.modules: list[Any] = []
 
         self.target_path = target_path
         self.results_path = results_path
@@ -65,10 +63,10 @@ class Command:
 
         # This list will contain all executed modules.
         # We can use this to reference e.g. self.executed[0].results.
-        self.executed: list[MVTModule] = []
+        self.executed: list[Any] = []
         self.hashes = hashes
         self.hash_values: list[dict[str, Any]] = []
-        self.timeline: ModuleTimeline = []
+        self.timeline: list[dict[str, Any]] = []
 
         # Load IOCs
         self._create_storage()
@@ -260,84 +258,20 @@ class Command:
         console.print("")
         console.print(panel)
 
-    def _ordered_modules(self) -> Optional[list[type[MVTModule]]]:
-        """Return enabled modules in stable topological order."""
-        module_indexes = {module: index for index, module in enumerate(self.modules)}
-
-        if self.module_name:
-            selected = [
-                module for module in self.modules if module.__name__ == self.module_name
-            ]
-        else:
-            selected = [module for module in self.modules if module.enabled]
-
-        required = set(selected)
-        pending = list(selected)
-        while pending:
-            module = pending.pop()
-            for dependency in module.dependencies:
-                if dependency not in module_indexes:
-                    self.log.warning(
-                        "Module %s depends on unavailable module %s. "
-                        "No modules will be run.",
-                        module.__name__,
-                        dependency.__name__,
-                    )
-                    return None
-                if dependency not in required:
-                    required.add(dependency)
-                    pending.append(dependency)
-
-        dependents: dict[type[MVTModule], list[type[MVTModule]]] = {
-            module: [] for module in required
-        }
-        indegree = {module: 0 for module in required}
-        for module in required:
-            for dependency in module.dependencies:
-                if dependency not in required:
-                    continue
-                dependents[dependency].append(module)
-                indegree[module] += 1
-
-        ready: list[tuple[int, type[MVTModule]]] = []
-        for module, count in indegree.items():
-            if count == 0:
-                heappush(ready, (module_indexes[module], module))
-
-        ordered = []
-        while ready:
-            _, module = heappop(ready)
-            ordered.append(module)
-            for dependent in dependents[module]:
-                indegree[dependent] -= 1
-                if indegree[dependent] == 0:
-                    heappush(ready, (module_indexes[dependent], dependent))
-
-        if len(ordered) != len(required):
-            cyclic_modules = sorted(
-                (module.__name__ for module, count in indegree.items() if count > 0)
-            )
-            self.log.warning(
-                "Circular module dependency detected involving: %s. "
-                "No modules will be run.",
-                ", ".join(cyclic_modules),
-            )
-            return None
-
-        return ordered
-
     def run(self) -> None:
-        ordered_modules = self._ordered_modules()
-        if ordered_modules is None:
-            return
-
         try:
             self.init()
         except NotImplementedError:
             pass
 
-        executed_by_type: dict[type[MVTModule], MVTModule] = {}
+        for module in self.modules:
-        for module in ordered_modules:
+            if self.module_name and module.__name__ != self.module_name:
+                continue
+
+            if not module.enabled and not (
+                self.module_name and module.__name__ == self.module_name
+            ):
+                continue
 
             module_logger = logging.getLogger(module.__module__)
 
@@ -347,10 +281,6 @@ class Command:
                 module_options=self.module_options,
                 log=module_logger,
             )
-            m.dependency_modules = {
-                dependency: executed_by_type[dependency]
-                for dependency in module.dependencies
-            }
 
             if self.iocs.total_ioc_count:
                 m.indicators = self.iocs
@@ -374,7 +304,6 @@ class Command:
                 return
 
             self.executed.append(m)
-            executed_by_type[module] = m
             self.timeline.extend(m.timeline)
             self.alertstore.extend(m.alertstore.alerts)
 

src/mvt/common/module.py

@@ -9,7 +9,7 @@ import logging
 import os
 import re
 from dataclasses import asdict, is_dataclass
-from typing import Any, Dict, Optional, Sequence
+from typing import Any, Dict, Optional
 
 from .alerts import AlertStore
 from .indicators import Indicators
@@ -43,7 +43,6 @@ class MVTModule:
 
     enabled: bool = True
     slug: Optional[str] = None
-    dependencies: Sequence[type["MVTModule"]] = ()
 
     def __init__(
         self,
@@ -72,7 +71,6 @@ class MVTModule:
         self.file_path: Optional[str] = file_path
         self.target_path: Optional[str] = target_path
         self.results_path: Optional[str] = results_path
-        self.serial: Optional[str] = None
         self.module_options: Dict[str, Any] = module_options if module_options else {}
 
         self.log = log
@@ -81,13 +79,6 @@ class MVTModule:
 
         self.results: ModuleResults = results if results is not None else []
         self.timeline: ModuleTimeline = []
-        self.dependency_modules: Dict[type["MVTModule"], "MVTModule"] = {}
-
-    def get_dependency_results(
-        self, module_class: type["MVTModule"]
-    ) -> ModuleResults:
-        """Return the results produced by a prerequisite module."""
-        return self.dependency_modules[module_class].results
 
     @classmethod
     def from_json(cls, json_path: str, log: logging.Logger):

tests/android/test_intrusion_logs.py

@@ -143,58 +143,6 @@ def test_check_intrusion_logs_parses_core_and_unknown_security_events(
     assert "Please open an issue on GitHub" in caplog.text
 
 
-def test_check_intrusion_logs_treats_event_id_as_security_event_metadata(
-    tmp_path, caplog
-):
-    _write_ndjson(
-        tmp_path / "intrusion.txt",
-        [
-            {
-                "security_event": {
-                    "event_id": 191,
-                    "event_time": 1_700_000_002_000_000_000,
-                    "keyguard_dismiss_auth_attempt": {
-                        "success": True,
-                        "method_strength": 0,
-                    },
-                }
-            },
-            {
-                "security_event": {
-                    "event_id": 192,
-                    "event_time": 1_700_000_003_000_000_000,
-                    "keyguard_dismissed": {},
-                }
-            },
-        ],
-    )
-
-    with caplog.at_level(logging.WARNING):
-        cmd = CmdAndroidCheckIntrusionLogs(target_path=str(tmp_path))
-        cmd.run()
-
-    security_module = next(
-        module for module in cmd.executed if isinstance(module, SecurityEvent)
-    )
-    assert security_module.event_type_counts == {
-        "keyguard_dismiss_auth_attempt": 1,
-        "keyguard_dismissed": 1,
-    }
-    assert [event["event_id"] for event in security_module.results] == [191, 192]
-
-    keyguard_events = {
-        event["event"]: event
-        for event in cmd.timeline
-        if event["event"]
-        in {"keyguard_dismiss_auth_attempt", "keyguard_dismissed"}
-    }
-    assert "Auth attempt: Success" in keyguard_events[
-        "keyguard_dismiss_auth_attempt"
-    ]["data"]
-    assert keyguard_events["keyguard_dismissed"]["data"] == "Keyguard dismissed"
-    assert "unknown intrusion logging security event type(s): event_id" not in caplog.text
-
-
 def test_check_intrusion_logs_cli_lists_modules(tmp_path):
     _write_ndjson(tmp_path / "intrusion.txt", [])
 

tests/common/test_command.py

@@ -4,59 +4,11 @@
 #   https://license.mvt.re/1.1/
 
 import json
-import logging
 
 from mvt.common.command import Command
-from mvt.common.module import MVTModule
-
-
-class RecordingModule(MVTModule):
-    run_order: list[str] = []
-
-    def run(self):
-        self.run_order.append(self.__class__.__name__)
-
-    def check_indicators(self):
-        pass
-
-
-class FirstModule(RecordingModule):
-    def run(self):
-        super().run()
-        self.results = ["first"]
-
-
-class SecondModule(RecordingModule):
-    dependencies = (FirstModule,)
-
-    def run(self):
-        super().run()
-        self.results = self.get_dependency_results(FirstModule) + ["second"]
-
-
-class ThirdModule(RecordingModule):
-    dependencies = (SecondModule,)
-
-
-class IndependentModule(RecordingModule):
-    pass
-
-
-class RecordingCommand(Command):
-    def init(self):
-        self.initialized = True
-
-    def module_init(self, module):
-        pass
-
-    def finish(self):
-        pass
 
 
 class TestCommand:
-    def setup_method(self):
-        RecordingModule.run_order = []
-
     def test_store_alerts_handles_bytes(self, tmp_path):
         cmd = Command(results_path=str(tmp_path))
         cmd.alertstore.medium(
@@ -69,66 +21,3 @@ class TestCommand:
 
         alerts = json.loads((tmp_path / "alerts.json").read_text())
         assert alerts[0]["event"]["payload"] == "\\xa8\\xa9"
-
-    def test_modules_run_in_stable_topological_order(self):
-        cmd = RecordingCommand()
-        cmd.modules = [ThirdModule, IndependentModule, SecondModule, FirstModule]
-
-        cmd.run()
-
-        assert RecordingModule.run_order == [
-            "IndependentModule",
-            "FirstModule",
-            "SecondModule",
-            "ThirdModule",
-        ]
-        second = next(module for module in cmd.executed if isinstance(module, SecondModule))
-        assert second.results == ["first", "second"]
-
-    def test_selected_module_runs_transitive_dependencies(self):
-        cmd = RecordingCommand(module_name="ThirdModule")
-        cmd.modules = [ThirdModule, SecondModule, FirstModule, IndependentModule]
-
-        cmd.run()
-
-        assert RecordingModule.run_order == [
-            "FirstModule",
-            "SecondModule",
-            "ThirdModule",
-        ]
-
-    def test_circular_dependency_warns_and_stops(self, caplog):
-        class CircularOne(RecordingModule):
-            pass
-
-        class CircularTwo(RecordingModule):
-            dependencies = (CircularOne,)
-
-        CircularOne.dependencies = (CircularTwo,)
-
-        cmd = RecordingCommand()
-        cmd.modules = [CircularOne, CircularTwo]
-
-        with caplog.at_level(logging.WARNING):
-            cmd.run()
-
-        assert RecordingModule.run_order == []
-        assert not hasattr(cmd, "initialized")
-        assert "Circular module dependency detected" in caplog.text
-
-    def test_unavailable_dependency_warns_and_stops(self, caplog):
-        class UnavailableModule(RecordingModule):
-            pass
-
-        class DependentModule(RecordingModule):
-            dependencies = (UnavailableModule,)
-
-        cmd = RecordingCommand()
-        cmd.modules = [DependentModule]
-
-        with caplog.at_level(logging.WARNING):
-            cmd.run()
-
-        assert RecordingModule.run_order == []
-        assert not hasattr(cmd, "initialized")
-        assert "depends on unavailable module UnavailableModule" in caplog.text