diff --git a/CHANGELOG.md b/CHANGELOG.md index 77cba765b0..bc311066c7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,7 @@ Please refer to the [NEWS](NEWS.md) for a list of changes which have an affect o - `intelmq.bots.parsers.generic_csv.parser_csv`: Handle empty string parameter `columns_required` as unset (PR#2680 by Sebastian Wagner, fixes #2679). #### Experts +- `intelmq.bots.experts.asn_lookup.expert`: Allow `--update-database` to create a missing database file (fixes #2689 by Haitao Zheng). #### Outputs - `intelmq.bots.outputs.smtp_batch.output`: diff --git a/NEWS.md b/NEWS.md index af70eeecc5..e4d5319870 100644 --- a/NEWS.md +++ b/NEWS.md @@ -16,6 +16,7 @@ Please refer to the change log for a full list of changes. ### Requirements ### Tools +- `intelmq.bots.experts.asn_lookup.expert --update-database` now creates the configured database file if it does not exist yet. ### Data Format diff --git a/intelmq/bots/experts/asn_lookup/expert.py b/intelmq/bots/experts/asn_lookup/expert.py index 008e291dc7..f126795b0d 100644 --- a/intelmq/bots/experts/asn_lookup/expert.py +++ b/intelmq/bots/experts/asn_lookup/expert.py @@ -123,9 +123,10 @@ def update_database(cls, verbose=False): raise MissingDependencyError("pyasn") for database_path in set(bots.values()): - if not Path(database_path).is_file(): - raise ValueError('Database file does not exist or is not a file.') - elif not os.access(database_path, os.W_OK): + database_file = Path(database_path) + if database_file.exists() and not database_file.is_file(): + raise ValueError('Database path exists but is not a file.') + elif database_file.exists() and not os.access(database_path, os.W_OK): raise ValueError('Database file is not writeable.') try: diff --git a/intelmq/tests/bots/experts/asn_lookup/test_expert.py b/intelmq/tests/bots/experts/asn_lookup/test_expert.py index 788f9269e4..9ff27e546a 100644 --- a/intelmq/tests/bots/experts/asn_lookup/test_expert.py +++ b/intelmq/tests/bots/experts/asn_lookup/test_expert.py @@ -8,6 +8,10 @@ """ import unittest +from pathlib import Path +from tempfile import TemporaryDirectory +from types import SimpleNamespace +from unittest import mock import pkg_resources @@ -62,5 +66,65 @@ def test_ipv6_lookup(self): self.assertMessageEqual(0, EXAMPLE_OUTPUT6) +class TestASNLookupUpdateDatabase(unittest.TestCase): + def test_update_database_creates_missing_database_file(self): + def dump_prefixes_to_file(prefixes, database_path): + Path(database_path).write_text("fake database", encoding="utf-8") + + class Session: + def get(self, url): + if url.endswith("/RIBS/"): + return SimpleNamespace( + text='rib', + status_code=200, + url=url, + ) + if url.endswith(".bz2"): + return SimpleNamespace(text="", content=b"BZh91AY&SY", status_code=200, url=url) + return SimpleNamespace( + text='2026.072026.06', + status_code=200, + url=url, + ) + + pyasn = SimpleNamespace( + mrtx=SimpleNamespace( + parse_mrt_file=mock.Mock(return_value=["prefixes"]), + dump_prefixes_to_file=mock.Mock(side_effect=dump_prefixes_to_file), + ) + ) + + with TemporaryDirectory() as tempdir: + database_path = Path(tempdir) / "asn_lookup" / "ipasn.dat" + runtime_config = { + "asn-lookup-expert": { + "module": "intelmq.bots.experts.asn_lookup.expert", + "parameters": {"database": str(database_path)}, + } + } + controller = mock.Mock() + + with ( + mock.patch( + "intelmq.bots.experts.asn_lookup.expert.get_bots_settings", + return_value=runtime_config, + ), + mock.patch( + "intelmq.bots.experts.asn_lookup.expert.create_request_session", + return_value=Session(), + ), + mock.patch("intelmq.bots.experts.asn_lookup.expert.pyasn", pyasn), + mock.patch( + "intelmq.bots.experts.asn_lookup.expert.IntelMQController", + return_value=controller, + ), + ): + ASNLookupExpertBot.update_database() + + self.assertTrue(database_path.is_file()) + pyasn.mrtx.dump_prefixes_to_file.assert_called_once_with(["prefixes"], str(database_path)) + controller.bot_reload.assert_called_once_with("asn-lookup-expert") + + if __name__ == '__main__': # pragma: no cover unittest.main()