asn: Check for database file existance and writability

fixes #2566

Author: sebix

CHANGELOG.md

@@ -31,6 +31,7 @@ Please refer to the [NEWS](NEWS.md) for a list of changes which have an affect o
 #### Experts
 - `intelmq.bots.experts.asn_lookup.expert`:
   - Print URLs to stdout only in verbose mode (PR#2591 by Sebastian Wagner).
+  - Check for database file existence and writability (fixes #2566).
   - Use database path matching to installation type (PR#2606 by Sebastian Wagner).
 - `intelmq.bots.experts.fake.expert`: Use database path matching to installation type (PR#2606 by Sebastian Wagner).
 

intelmq/bots/experts/asn_lookup/expert.py

@@ -8,8 +8,8 @@
 import re
 import sys
 import bz2
-import pathlib
 import requests
+from pathlib import Path
 
 from intelmq import VAR_STATE_PATH
 from intelmq.lib.bot import ExpertBot
@@ -41,6 +41,8 @@ def init(self):
             self.logger.error("Read 'bots/experts/asn_lookup/README' and "
                               "follow the procedure.")
             self.stop()
+        if not Path(self.database).is_file:
+            raise ValueError('Database file does not exist or is not a file.')
 
     def process(self):
         event = self.receive_message()
@@ -113,6 +115,12 @@ def update_database(cls, verbose=False):
         if pyasn is None:
             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):
+                raise ValueError('Database file is not writeable.')
+
         try:
             if verbose:
                 print("Searching for the latest database update...")
@@ -160,7 +168,7 @@ def update_database(cls, verbose=False):
             prefixes = pyasn.mrtx.parse_mrt_file(archive, print_progress=False, skip_record_on_error=True)
 
         for database_path in set(bots.values()):
-            database_dir = pathlib.Path(database_path).parent
+            database_dir = Path(database_path).parent
             database_dir.mkdir(parents=True, exist_ok=True)
             pyasn.mrtx.dump_prefixes_to_file(prefixes, database_path)