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()