Handle bracketed URL netloc sanitization

intelmq/lib/harmonization.py

@@ -1084,6 +1084,15 @@ class URL(String):
     Valid values must have the host (network location part).
     """
 
+    @staticmethod
+    def _has_netloc(value: str) -> bool:
+        try:
+            result = parse.urlsplit(value)
+        except ValueError:
+            match = re.match(r'^[A-Za-z][A-Za-z0-9+.-]*://([^/?#]+)', value)
+            return bool(match and match.group(1))
+        return result.netloc != ""
+
     @staticmethod
     def is_valid(value: str, sanitize: bool = False) -> bool:
         if sanitize:
@@ -1095,8 +1104,7 @@ def is_valid(value: str, sanitize: bool = False) -> bool:
         if value[0] in string.whitespace:
             return False
 
-        result = parse.urlsplit(value)
-        if result.netloc == "":
+        if not URL._has_netloc(value):
             return False
 
         return True
@@ -1110,15 +1118,18 @@ def sanitize(value: str) -> Optional[str]:
         value = value.replace('hxxp://', 'http://')
         value = value.replace('hxxps://', 'https://')
 
-        result = parse.urlsplit(value)
-        if result.scheme == "file" and result.netloc == '':
-            # add localhost as netloc
-            result_split = list(result)
-            result_split[1] = 'localhost'
-            value = parse.urlunsplit(result_split)
+        try:
             result = parse.urlsplit(value)
+        except ValueError:
+            return value if URL._has_netloc(value) else None
+        else:
+            if result.scheme == "file" and result.netloc == '':
+                # add localhost as netloc
+                result_split = list(result)
+                result_split[1] = 'localhost'
+                value = parse.urlunsplit(result_split)
 
-        if result.netloc != "":
+        if URL._has_netloc(value):
             return value
 
     @staticmethod

intelmq/tests/bots/parsers/html_table/test_parser_column_split.py

@@ -70,7 +70,6 @@ def test_event_with_split(self):
         self.run_bot()
         self.assertMessageEqual(0, EXAMPLE_EVENT)
 
-    @unittest.skip("Change in urllib prevent invalid URLs to be processed, see #2377")
     def test_event_without_split(self):
         self.sysconfig = {"columns": ["time.source", "source.url", "malware.hash.md5",
                                       "source.ip", "__IGNORE__"],

intelmq/tests/lib/test_harmonization.py

@@ -407,6 +407,7 @@ def test_url_valid(self):
         self.assertTrue(harmonization.URL.is_valid('http://example.com'))
         self.assertTrue(harmonization.URL.is_valid('http://example.com/foo'))
         self.assertTrue(harmonization.URL.is_valid('file://localhost/etc/hosts'))
+        self.assertTrue(harmonization.URL.is_valid('http://[D] example.com/foo'))
 
     def test_url_invalid(self):
         """ Test URL.is_valid with invalid arguments. """
@@ -426,6 +427,8 @@ def test_url_sanitize(self):
                                                    sanitize=True))
         self.assertEqual(harmonization.URL.sanitize(' http://example.com'),
                          'http://example.com')
+        self.assertEqual(harmonization.URL.sanitize('http://[D] example.com/foo'),
+                         'http://[D] example.com/foo')
 
     def test_url_sanitize_invalid(self):
         """ Test URL.is_valid with valid arguments. """