Merge pull request #1588 from funilrys/support-raw

Introduction of the support of RAW lines/sources

setup.cfg

@@ -1,2 +1,6 @@
 [flake8]
 max-line-length = 120
+extend-ignore = E203
+
+[isort]
+profile = black

testUpdateHostsFile.py

@@ -633,6 +633,30 @@ class TestMatchesExclusions(Base):
         ]:
             self.assertTrue(matches_exclusions(domain, exclusion_regexes))
 
+    def test_match_raw_list(self):
+        exclusion_regexes = [r".*\.com", r".*\.org", r".*\.edu"]
+        exclusion_regexes = [re.compile(regex) for regex in exclusion_regexes]
+
+        for domain in [
+            "hulu.com",
+            "yahoo.com",
+            "adaway.org",
+            "education.edu",
+        ]:
+            self.assertTrue(matches_exclusions(domain, exclusion_regexes))
+
+    def test_no_match_raw_list(self):
+        exclusion_regexes = [r".*\.org", r".*\.edu"]
+        exclusion_regexes = [re.compile(regex) for regex in exclusion_regexes]
+
+        for domain in [
+            "localhost",
+            "hulu.com",
+            "yahoo.com",
+            "cloudfront.net",
+        ]:
+            self.assertFalse(matches_exclusions(domain, exclusion_regexes))
+
 
 # End Exclusion Logic
 
@@ -806,13 +830,11 @@ class TestNormalizeRule(BaseStdout):
     def test_no_match(self):
         kwargs = dict(target_ip="0.0.0.0", keep_domain_comments=False)
 
+        # Note: "Bare"- Domains are accepted. IP are excluded.
         for rule in [
-            "foo",
             "128.0.0.1",
-            "bar.com/usa",
             "0.0.0 google",
             "0.1.2.3.4 foo/bar",
-            "twitter.com",
         ]:
             self.assertEqual(normalize_rule(rule, **kwargs), (None, None))
 
@@ -874,13 +896,43 @@ class TestNormalizeRule(BaseStdout):
 
             sys.stdout = StringIO()
 
+    def test_no_comment_raw(self):
+        for rule in ("twitter.com", "google.com", "foo.bar.edu"):
+            expected = (rule, "0.0.0.0 " + rule + "\n")
 
-class TestStripRule(Base):
-    def test_strip_empty(self):
-        for line in ["0.0.0.0", "domain.com", "foo"]:
-            output = strip_rule(line)
+            actual = normalize_rule(
+                rule, target_ip="0.0.0.0", keep_domain_comments=False
+            )
+            self.assertEqual(actual, expected)
+
+            # Nothing gets printed if there's a match.
+            output = sys.stdout.getvalue()
             self.assertEqual(output, "")
 
+            sys.stdout = StringIO()
+
+    def test_with_comments_raw(self):
+        for target_ip in ("0.0.0.0", "127.0.0.1", "8.8.8.8"):
+            for comment in ("foo", "bar", "baz"):
+                rule = "1.google.co.uk " + comment
+                expected = (
+                    "1.google.co.uk",
+                    (str(target_ip) + " 1.google.co.uk # " + comment + "\n"),
+                )
+
+                actual = normalize_rule(
+                    rule, target_ip=target_ip, keep_domain_comments=True
+                )
+                self.assertEqual(actual, expected)
+
+                # Nothing gets printed if there's a match.
+                output = sys.stdout.getvalue()
+                self.assertEqual(output, "")
+
+                sys.stdout = StringIO()
+
+
+class TestStripRule(Base):
     def test_strip_exactly_two(self):
         for line in [
             "0.0.0.0 twitter.com",
@@ -903,6 +955,28 @@ class TestStripRule(Base):
             output = strip_rule(line + comment)
             self.assertEqual(output, line + comment)
 
+    def test_strip_raw(self):
+        for line in [
+            "twitter.com",
+            "facebook.com",
+            "google.com",
+            "foo.bar.edu",
+        ]:
+            output = strip_rule(line)
+            self.assertEqual(output, line)
+
+    def test_strip_raw_with_comment(self):
+        comment = " # comments here galore"
+
+        for line in [
+            "twitter.com",
+            "facebook.com",
+            "google.com",
+            "foo.bar.edu",
+        ]:
+            output = strip_rule(line + comment)
+            self.assertEqual(output, line + comment)
+
 
 class TestWriteOpeningHeader(BaseMockDir):
     def setUp(self):
@@ -1572,20 +1646,30 @@ class GetFileByUrl(BaseStdout):
         self.assertEqual(expected, actual)
 
     def test_connect_unknown_domain(self):
-        test_url = "http://doesnotexist.google.com"  # leads to exception: ConnectionError
-        with mock.patch("requests.get", side_effect=requests.exceptions.ConnectionError):
+        test_url = (
+            "http://doesnotexist.google.com"  # leads to exception: ConnectionError
+        )
+        with mock.patch(
+            "requests.get", side_effect=requests.exceptions.ConnectionError
+        ):
             return_value = get_file_by_url(test_url)
         self.assertIsNone(return_value)
         printed_output = sys.stdout.getvalue()
-        self.assertEqual(printed_output, "Error retrieving data from {}\n".format(test_url))
+        self.assertEqual(
+            printed_output, "Error retrieving data from {}\n".format(test_url)
+        )
 
     def test_invalid_url(self):
         test_url = "http://fe80::5054:ff:fe5a:fc0"  # leads to exception: InvalidURL
-        with mock.patch("requests.get", side_effect=requests.exceptions.ConnectionError):
+        with mock.patch(
+            "requests.get", side_effect=requests.exceptions.ConnectionError
+        ):
             return_value = get_file_by_url(test_url)
         self.assertIsNone(return_value)
         printed_output = sys.stdout.getvalue()
-        self.assertEqual(printed_output, "Error retrieving data from {}\n".format(test_url))
+        self.assertEqual(
+            printed_output, "Error retrieving data from {}\n".format(test_url)
+        )
 
 
 class TestWriteData(Base):

updateHostsFile.py

@@ -20,6 +20,7 @@ import sys
 import tempfile
 import time
 from glob import glob
+from typing import Optional, Tuple
 
 # Detecting Python 3 for version-dependent implementations
 PY3 = sys.version_info >= (3, 0)
@@ -31,8 +32,10 @@ if not PY3:
 try:
     import requests
 except ImportError:
-    raise ImportError("This project's dependencies have changed. The Requests library ("
-                      "https://requests.readthedocs.io/en/master/) is now required.")
+    raise ImportError(
+        "This project's dependencies have changed. The Requests library ("
+        "https://requests.readthedocs.io/en/master/) is now required."
+    )
 
 
 # Syntactic sugar for "sudo" command in UNIX / Linux
@@ -629,7 +632,11 @@ def matches_exclusions(stripped_rule, exclusion_regexes):
         Whether or not the rule string matches a provided exclusion.
     """
 
-    stripped_domain = stripped_rule.split()[1]
+    try:
+        stripped_domain = stripped_rule.split()[1]
+    except IndexError:
+        # Example: 'example.org' instead of '0.0.0.0 example.org'
+        stripped_domain = stripped_rule
 
     for exclusionRegex in exclusion_regexes:
         if exclusionRegex.search(stripped_domain):
@@ -981,6 +988,37 @@ def normalize_rule(rule, target_ip, keep_domain_comments):
         and spacing reformatted.
     """
 
+    def normalize_response(
+        extracted_hostname: str, extracted_suffix: Optional[str]
+    ) -> Tuple[str, str]:
+        """
+        Normalizes the responses after the provision of the extracted
+        hostname and suffix - if exist.
+
+        Parameters
+        ----------
+        extracted_hostname: str
+            The extracted hostname to work with.
+        extracted_suffix: str
+            The extracted suffix to with.
+
+        Returns
+        -------
+        normalized_response: tuple
+            A tuple of the hostname and the rule string with spelling
+            and spacing reformatted.
+        """
+
+        rule = "%s %s" % (target_ip, extracted_hostname)
+
+        if keep_domain_comments and extracted_suffix:
+            if not extracted_suffix.strip().startswith("#"):
+                rule += " #%s" % extracted_suffix
+            else:
+                rule += " %s" % extracted_suffix
+
+        return extracted_hostname, rule + "\n"
+
     """
     first try: IP followed by domain
     """
@@ -992,15 +1030,8 @@ def normalize_rule(rule, target_ip, keep_domain_comments):
 
         # Explicitly lowercase and trim the hostname.
         hostname = hostname.lower().strip()
-        rule = "%s %s" % (target_ip, hostname)
 
-        if suffix and keep_domain_comments:
-            if not suffix.strip().startswith("#"):
-                rule += " #%s" % suffix
-            else:
-                rule += " %s" % suffix
-
-        return hostname, rule + "\n"
+        return normalize_response(hostname, suffix)
 
     """
     next try: IP address followed by host IP address
@@ -1012,15 +1043,22 @@ def normalize_rule(rule, target_ip, keep_domain_comments):
         ip_host, suffix = result.group(2, 3)
         # Explicitly trim the ip host.
         ip_host = ip_host.strip()
-        rule = "%s %s" % (target_ip, ip_host)
 
-        if suffix and keep_domain_comments:
-            if not suffix.strip().startswith("#"):
-                rule += " #%s" % suffix
-            else:
-                rule += " %s" % suffix
+        return normalize_response(ip_host, suffix)
 
-        return ip_host, rule + "\n"
+    """
+    next try: Keep RAW domain.
+    """
+    regex = r"^\s*([\w\.-]+[a-zA-Z])(.*)"
+    result = re.search(regex, rule)
+
+    if result:
+        hostname, suffix = result.group(1, 2)
+
+        # Explicitly lowercase and trim the hostname.
+        hostname = hostname.lower().strip()
+
+        return normalize_response(hostname, suffix)
 
     """
     finally, if we get here, just belch to screen
@@ -1044,12 +1082,7 @@ def strip_rule(line):
         The sanitized rule.
     """
 
-    split_line = line.split()
-    if len(split_line) < 2:
-        # just return blank
-        return ""
-    else:
-        return " ".join(split_line)
+    return " ".join(line.split())
 
 
 def write_opening_header(final_file, **header_params):
@@ -1313,8 +1346,9 @@ def flush_dns_cache():
                 )
 
                 if os.path.isfile(service_file):
-                    if 0 != subprocess.call([systemctl, "status", service],
-                                            stdout=subprocess.DEVNULL):
+                    if 0 != subprocess.call(
+                        [systemctl, "status", service], stdout=subprocess.DEVNULL
+                    ):
                         continue
                     dns_cache_found = True