Defer indicator loading until first use (#919)

* Defer indicator loading until first use

* Load indicators once at the start of a run

The lazy property loads indicators wherever they are first read, which
in Command.run() is inside the module loop, after init(). For
check-androidqf that means the whole acquisition is walked before a
missing --iocs file is reported, and the STIX parsing lines land between
the module list and the first module.

Read the indicators once after the module list is settled, so that a bad
--module name still loads nothing, and hand the same object to every
module. check-iocs reads them once too, so that a wrong --iocs path is
reported even when no stored result matches a module.

* Drop two test assertions the change does not need

Retrying after a failed indicator load is incidental to the property
rather than a requirement, so nothing should pin it. The exact wording
of the check-backup rejection belongs to that command's own tests; exit
code 1 already shows the path was rejected before indicators were
loaded.

* Create the output folder and command.log when a run starts

Command.__init__ created the --output folder and attached the
command.log handler, so `--list-modules -o out` or a rejected target
path left an empty folder behind for a run that never happened.

Do both at the top of run(), before the module list is resolved so that
its warnings still reach the log. The nested commands run by
check-androidqf re-attach the handler at the start of their runs, as
they did at construction. Lines the CLI logs between construction and
run(), such as the target path being checked, no longer reach
command.log; info.json records the target path.

This is the remaining part of #888.

* Cache the indicators with functools.cached_property

A property with a setter and a backing attribute does by hand what
cached_property does: compute on first read, store the result on the
instance, and accept assignment, which is how nested commands receive
their parent's indicators. A failed load is still not cached.

* Announce the target from the command so that it reaches command.log

The CLI logged which backup, filesystem or acquisition it was about to
check just before run(), which is now before command.log exists, so the
line only reached the console. Each command logs it from init() instead,
which runs once the log is attached, and run() logs the module list
after init() so that the target still comes first. Nested commands
without a target path log nothing, as before.

* Log the Android backup path from the typed local

mypy cannot determine the type of target_path in this command, which the
surrounding lines already silence, so read the announced path from the
local that carries the type instead of adding another ignore.

---------

Co-authored-by: bitmeta69 <206962233+bitmeta69@users.noreply.github.com>
Co-authored-by: besendorf <janik@besendorf.org>
Co-authored-by: Donncha Ó Cearbhaill <donncha.ocearbhaill@amnesty.org>
This commit is contained in:
Daniel Conor Sullivan
2026-09-08 01:01:02 +01:00
committed by GitHub
co-authored by bitmeta69 besendorf Donncha Ó Cearbhaill
parent 60e652bafc
commit 3623e430f4
13 changed files with 137 additions and 25 deletions
+85
View File
@@ -5,8 +5,13 @@
import json
import logging
from unittest.mock import patch
import pytest
from click.testing import CliRunner
from mvt.common.command import Command
from mvt.common.indicators import Indicators
from mvt.common.module import MVTModule
@@ -197,6 +202,86 @@ class RecordingCommand(Command):
class TestCommand:
def test_listing_modules_does_not_load_indicators(self):
with patch("mvt.common.command.Indicators.load_indicators_files") as load:
cmd = RecordingCommand()
cmd.list_modules()
load.assert_not_called()
def test_output_folder_is_created_by_a_run_not_by_listing(self, tmp_path):
output_path = tmp_path / "out"
with patch("mvt.common.command.Indicators.load_indicators_files"):
cmd = RecordingCommand(results_path=str(output_path))
cmd.list_modules()
assert not output_path.exists()
cmd.run()
assert (output_path / "command.log").is_file()
def test_indicators_load_once_and_are_shared(self, indicator_file, monkeypatch):
from mvt.common.config import settings
monkeypatch.setattr(settings, "STIX2", "")
monkeypatch.setattr(Indicators, "_load_downloaded_indicators", lambda self: None)
original = Indicators.load_indicators_files
with patch.object(
Indicators, "load_indicators_files", autospec=True, side_effect=original
) as load:
cmd = RecordingCommand(ioc_files=[indicator_file])
load.assert_not_called()
indicators = cmd.iocs
assert indicators.total_ioc_count == 9
assert len(indicators.ioc_collections) == 1
assert cmd.iocs is indicators
child = RecordingCommand(iocs=indicators)
assert child.iocs is indicators
load.assert_called_once_with(indicators, [indicator_file])
@pytest.mark.parametrize("assign", [False, True])
def test_supplied_empty_indicators_are_not_loaded(self, assign):
indicators = Indicators(logging.getLogger(__name__))
with patch.object(Indicators, "load_indicators_files") as load:
cmd = RecordingCommand(iocs=None if assign else indicators)
if assign:
cmd.iocs = indicators
assert cmd.iocs is indicators
assert cmd.iocs.total_ioc_count == 0
load.assert_not_called()
@pytest.mark.parametrize("list_modules", [False, True])
def test_backup_cli_does_not_load_indicators_before_analysis(
self, tmp_path, list_modules
):
from mvt.ios.cli import check_backup
args = [str(tmp_path)]
if list_modules:
args.insert(0, "--list-modules")
with patch.object(Indicators, "load_indicators_files") as load:
result = CliRunner().invoke(check_backup, args)
assert result.exit_code == (0 if list_modules else 1)
load.assert_not_called()
def test_run_checks_synthetic_indicators(self, indicator_file, monkeypatch):
from mvt.common.config import settings
monkeypatch.setattr(settings, "STIX2", "")
monkeypatch.setattr(Indicators, "_load_downloaded_indicators", lambda self: None)
class MatchingModule(RecordingModule):
def run(self):
self.results = ["https://example.org/test"]
def check_indicators(self):
self.detected = [
url for url in self.results if self.indicators.check_domain(url)
]
cmd = RecordingCommand(ioc_files=[indicator_file])
cmd.modules = [MatchingModule]
cmd.run()
assert cmd.executed[0].detected == ["https://example.org/test"]
assert cmd.executed[0].indicators is cmd.iocs
def setup_method(self):
RecordingModule.run_order = []