FIX: Possible race condition in file collector

Locking the file with python fcntl to aquire a file lock,
it may not support windows installations :/

Fixes #2128
Fixes #1631

Signed-off-by: Sebastian Waldbauer <waldbauer@cert.at>

Author: waldbauer-certat

intelmq/bots/collectors/file/collector_file.py

@@ -24,6 +24,7 @@
 
 import fnmatch
 import os
+import fcntl
 
 import intelmq.lib.exceptions as exceptions
 from intelmq.lib.bot import CollectorBot
@@ -71,10 +72,15 @@ def process(self):
                         template.add("feed.url", "file://localhost%s" % filename)
                         template.add("extra.file_name", f)
 
-                        with open(filename, 'rb') as fh:
-                            for report in generate_reports(template, fh, self.chunk_size,
-                                                           self.chunk_replicate_header):
-                                self.send_message(report)
+                        try:
+                            with open(filename, 'rb') as fh:
+                                fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                                for report in generate_reports(template, fh,
+                                                                self.chunk_size, self.chunk_replicate_header):
+                                    self.send_message(report)
+                                fcntl.flock(fh, fcntl.LOCK_UN)
+                        except BlockingIOError:
+                            self.logger.info("File is already being used by another process, skipping.")
 
                         if self.delete_file:
                             try:

intelmq/tests/bots/collectors/file/test_collector.py

@@ -8,6 +8,7 @@
 """
 import os
 import unittest
+import fcntl
 
 import intelmq.lib.test as test
 import intelmq.lib.utils as utils
@@ -49,6 +50,16 @@ def test_events(self):
 
         self.assertMessageEqual(0, OUTPUT)
 
+    def test_file_lock(self):
+        f = open(PATH, 'rb')
+        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        self.run_bot(iterations=1)
+        self.assertLogMatches('File is already being used by another process, skipping.', levelname="INFO")
+        fcntl.flock(f, fcntl.LOCK_UN)
+        f.close()
+        self.run_bot(iterations=1)
+        self.assertMessageEqual(0, OUTPUT)
+
 
 if __name__ == '__main__':  # pragma: no cover
     unittest.main()