mirror of
https://github.com/mvt-project/mvt.git
synced 2026-08-16 16:10:25 +02:00
Store message URLs in analysis output (#856)
This commit is contained in:
@@ -42,6 +42,15 @@ 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"),)
|
||||
|
||||
@@ -87,6 +96,21 @@ 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]
|
||||
|
||||
@@ -197,6 +197,15 @@ class TestIndicators:
|
||||
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",
|
||||
|
||||
@@ -18,6 +18,14 @@ 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):
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
# 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",
|
||||
}
|
||||
]
|
||||
@@ -33,6 +33,24 @@ class TestCheckAndroidqfCommand:
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user