BUG: utils/load_configuration: fix loading JSON with tabs

intelmq/lib/utils.py

@@ -45,6 +45,7 @@
 from dateutil.relativedelta import relativedelta
 from termstyle import red
 from ruamel.yaml import YAML
+from ruamel.yaml.scanner import ScannerError
 
 import intelmq
 from intelmq.lib.exceptions import DecodingError
@@ -213,7 +214,13 @@ def load_configuration(configuration_filepath: str) -> dict:
     """
     if os.path.exists(configuration_filepath):
         with open(configuration_filepath, 'r') as fpconfig:
-            config = yaml.load(fpconfig)
+            try:
+                config = yaml.load(fpconfig)
+            except ScannerError as exc:
+                if "found character '\\t' that cannot start any token" in exc.problem:
+                    fpconfig.seek(0)
+                    return json.load(fpconfig)
+                raise
     else:
         raise ValueError('File not found: %r.' % configuration_filepath)
     return config

intelmq/tests/assets/example-invalid.yaml

@@ -0,0 +1,2 @@
+some_string: Hello World!
+other_string: with a : in it

intelmq/tests/assets/example-invalid.yaml.license

@@ -0,0 +1,2 @@
+SPDX-FileCopyrightText: 2018 Sebastian Wagner
+SPDX-License-Identifier: AGPL-3.0-or-later

intelmq/tests/assets/example.yaml

@@ -0,0 +1,10 @@
+some_string: Hello World!
+other_string: "with a : in it"
+now more:
+  - values
+  - in
+  - a
+  - list
+other: true
+types: -4
+final: 0.5

intelmq/tests/assets/example.yaml.license

@@ -0,0 +1,2 @@
+SPDX-FileCopyrightText: 2018 Sebastian Wagner
+SPDX-License-Identifier: AGPL-3.0-or-later

intelmq/tests/assets/foobar.json

@@ -0,0 +1,3 @@
+{
+    "foo": "bar"
+}

intelmq/tests/assets/foobar.json.license

@@ -0,0 +1,2 @@
+SPDX-FileCopyrightText: 2018 Sebastian Wagner
+SPDX-License-Identifier: AGPL-3.0-or-later

intelmq/tests/assets/tab-whitespace.json

@@ -0,0 +1,3 @@
+{
+	"foo": "bar"
+}

intelmq/tests/assets/tab-whitespace.json.license

@@ -0,0 +1,2 @@
+SPDX-FileCopyrightText: 2018 Sebastian Wagner
+SPDX-License-Identifier: AGPL-3.0-or-later

intelmq/tests/lib/test_utils.py

@@ -25,6 +25,7 @@
 import json
 
 import termstyle
+from ruamel.yaml.scanner import ScannerError
 
 from intelmq.tests.test_conf import CerberusTests
 import intelmq.lib.utils as utils
@@ -186,7 +187,7 @@ def test_parse_logline_syslog(self):
         line = ("Feb 22 10:17:10 host malware-domain-list-collector: ERROR "
                 "Something went wrong")
         thread = ("Feb 22 10:17:10 host malware-domain-list-collector.4: ERROR "
-                "Something went wrong")
+                  "Something went wrong")
 
         actual = utils.parse_logline(line, regex=utils.SYSLOG_REGEX)
         self.assertEqual({'bot_id': 'malware-domain-list-collector',
@@ -320,6 +321,36 @@ def test_get_global_settings(self):
         self.assertEqual(defaults['http_proxy'], 'http://localhost:8080')
         self.assertEqual(defaults['https_proxy'], 'http://localhost:8080')
 
+    def test_load_configuration_json(self):
+        """ Test load_configuration with a JSON file containing space whitespace """
+        filename = os.path.join(os.path.dirname(__file__), '../assets/foobar.json')
+        self.assertEqual(utils.load_configuration(filename), {'foo': 'bar'})
+
+    def test_load_configuration_json_tabs(self):
+        """ Test load_configuration with a JSON file containing tab whitespace """
+        filename = os.path.join(os.path.dirname(__file__), '../assets/tab-whitespace.json')
+        self.assertEqual(utils.load_configuration(filename), {'foo': 'bar'})
+
+    def test_load_configuration_yaml(self):
+        """ Test load_configuration with a YAML file """
+        filename = os.path.join(os.path.dirname(__file__), '../assets/example.yaml')
+        self.assertEqual(utils.load_configuration(filename),
+                         {
+                            'some_string': 'Hello World!',
+                            'other_string': 'with a : in it',
+                            'now more': ['values', 'in', 'a', 'list'],
+                            'types': -4,
+                            'other': True,
+                            'final': 0.5,
+                        }
+                        )
+
+    def test_load_configuration_yaml_invalid(self):
+        """ Test load_configuration with an invalid YAML file """
+        filename = os.path.join(os.path.dirname(__file__), '../assets/example-invalid.yaml')
+        with self.assertRaises(ScannerError):
+            utils.load_configuration(filename)
+
 
 if __name__ == '__main__':  # pragma: no cover
     unittest.main()