filter mode http

Author: 6C656C65

config.ini.example

@@ -14,6 +14,7 @@ html_403 = assets/403.html
 
 [Filtering]
 no_filter = false
+filter_mode = local
 blocked_sites = config/blocked_sites.txt
 blocked_url = config/blocked_url.txt
 

pyproxy.py

@@ -49,6 +49,12 @@
         help="Path to the custom 403 Forbidden HTML page"
     )
     parser.add_argument("--no-filter", action="store_true", help="Disable URL and domain filtering")
+    parser.add_argument(
+        "--filter-mode",
+        type=str,
+        choices=["local", "http"],
+        help="Filter list mode"
+    )
     parser.add_argument(
         "--blocked-sites",
         type=str,
@@ -97,6 +103,11 @@
         if args.no_filter
         else config.getboolean('Filtering', 'no_filter', fallback=False)
     )
+    filter_mode = (
+        args.filter_mode
+        if args.filter_mode
+        else config.get('Filtering', 'filter_mode', fallback="local")
+    )
     blocked_sites = (
         args.blocked_sites
         if args.blocked_sites
@@ -146,6 +157,7 @@
         block_log=block_log,
         html_403=html_403,
         no_filter=no_filter,
+        filter_mode=filter_mode,
         no_logging_access=no_logging_access,
         no_logging_block=no_logging_block,
         ssl_inspect=ssl_inspect,

tests/test_filter.py

@@ -33,7 +33,11 @@ def test_load_blacklist(self, mock_file):
         - Ensures that the sites and URLs are correctly read and loaded into sets.
         - Verifies that the function handles the file content properly.
         """
-        blocked_sites, blocked_urls = load_blacklist("blocked_sites.txt", "blocked_urls.txt")
+        blocked_sites, blocked_urls = load_blacklist(
+            "blocked_sites.txt",
+            "blocked_urls.txt",
+            "local"
+        )
 
         self.assertIn("blocked.com", blocked_sites)
         self.assertIn("allowed.com/blocked", blocked_sites)

utils/filter.py

@@ -14,33 +14,55 @@
 import time
 import sys
 import threading
+import requests
 
-def load_blacklist(blocked_sites_path: str, blocked_url_path: str) -> set:
+def load_blacklist(blocked_sites_path: str, blocked_url_path: str, filter_mode: str) -> set:
     """
-    Loads blocked FQDNs or URLs from a file into a set for fast lookup.
+    Loads blocked FQDNs or URLs from a file or URL into a set for fast lookup.
     
     Args:
-        blocked_sites_path (str): The path to the file containing blocked FQDNs.
-        blocked_url_path (str): The path to the file containing blocked URLs.
+        blocked_sites_path (str): The path or URL to the file containing blocked FQDNs.
+        blocked_url_path (str): The path or URL to the file containing blocked URLs.
+        filter_mode (str): Mode to determine if we load from local file or HTTP URL.
     
     Returns:
         set: A set of blocked domains/URLs.
     """
     blocked_sites = set()
     blocked_url = set()
 
-    with open(blocked_sites_path, 'r', encoding='utf-8') as f:
-        for line in f:
-            blocked_sites.add(line.strip())
-    with open(blocked_url_path, 'r', encoding='utf-8') as f:
-        for line in f:
-            blocked_url.add(line.strip())
+    def load_from_file(file_path: str) -> set:
+        data = set()
+        with open(file_path, 'r', encoding='utf-8') as f:
+            for line in f:
+                data.add(line.strip())
+        return data
+
+    def load_from_http(url: str) -> set:
+        data = set()
+        try:
+            response = requests.get(url, timeout=3)
+            response.raise_for_status()
+            for line in response.text.splitlines():
+                data.add(line.strip())
+        except requests.exceptions.RequestException as e:
+            raise requests.exceptions.RequestException(f"Failed to load data from {url}: {e}")
+        return data
+
+    if filter_mode == "local":
+        blocked_sites = load_from_file(blocked_sites_path)
+        blocked_url = load_from_file(blocked_url_path)
+    elif filter_mode == "http":
+        blocked_sites = load_from_http(blocked_sites_path)
+        blocked_url = load_from_http(blocked_url_path)
 
     return blocked_sites, blocked_url
 
+# pylint: disable=too-many-locals
 def filter_process(
     queue: multiprocessing.Queue,
     result_queue: multiprocessing.Queue,
+    filter_mode: str,
     blocked_sites_path: str,
     blocked_url_path: str
 ) -> None:
@@ -51,13 +73,14 @@ def filter_process(
         queue (multiprocessing.Queue): A queue to receive URL/domain for checking.
         result_queue (multiprocessing.Queue): A queue to send back the result of
                 the filtering (blocked or allowed).
+        filter_mode (str): Filter list mode (local or http).
         blocked_sites_path (str): The path to the file containing blocked FQDNs.
         blocked_url_path (str): The path to the file containing blocked URLs.
     """
     manager = multiprocessing.Manager()
     blocked_data = manager.dict({
-        "sites": load_blacklist(blocked_sites_path, blocked_url_path)[0],
-        "urls": load_blacklist(blocked_sites_path, blocked_url_path)[1],
+        "sites": load_blacklist(blocked_sites_path, blocked_url_path, filter_mode)[0],
+        "urls": load_blacklist(blocked_sites_path, blocked_url_path, filter_mode)[1],
     })
 
     error_event = threading.Event()
@@ -67,7 +90,8 @@ def file_monitor() -> None:
             while True:
                 new_blocked_sites, new_blocked_url = load_blacklist(
                     blocked_sites_path,
-                    blocked_url_path
+                    blocked_url_path,
+                    filter_mode
                 )
 
                 blocked_data["sites"] = new_blocked_sites

utils/proxy.py

@@ -32,7 +32,7 @@ class ProxyServer:
     """
     # pylint: disable=too-many-locals
     def __init__(self, host, port, debug, access_log, block_log,
-                 html_403, no_filter, no_logging_access, no_logging_block, ssl_inspect,
+                 html_403, no_filter, filter_mode, no_logging_access, no_logging_block, ssl_inspect,
                  blocked_sites, blocked_url, inspect_ca_cert, inspect_ca_key, inspect_certs_folder):
         """
         Initializes the ProxyServer instance with the provided configurations.
@@ -41,6 +41,7 @@ def __init__(self, host, port, debug, access_log, block_log,
         self.debug = debug
         self.html_403 = html_403
         self.no_filter = no_filter
+        self.filter_mode = filter_mode
         self.no_logging_access = no_logging_access
         self.no_logging_block = no_logging_block
         self.ssl_inspect = ssl_inspect
@@ -58,6 +59,7 @@ def __init__(self, host, port, debug, access_log, block_log,
         if not self.no_logging_block:
             self.block_logger = configure_file_logger(block_log, "BlockLogger")
 
+    # pylint: disable=too-many-branches, too-many-statements
     def start(self):
         """
         Starts the proxy server, initializes the filtering process if enabled,
@@ -71,6 +73,7 @@ def start(self):
             self.console_logger.debug("[*] debug = %s", self.debug)
             self.console_logger.debug("[*] html_403 = %s", self.html_403)
             self.console_logger.debug("[*] no_filter = %s", self.no_filter)
+            self.console_logger.debug("[*] filter_mode = %s", self.filter_mode)
             self.console_logger.debug("[*] no_logging_access = %s", self.no_logging_access)
             self.console_logger.debug("[*] no_logging_block = %s", self.no_logging_block)
             self.console_logger.debug("[*] ssl_inspect = %s", self.ssl_inspect)
@@ -101,19 +104,21 @@ def start(self):
                     except OSError as e:
                         self.console_logger.debug("OS error deleting %s: %s", file_path, e)
 
-        if not os.path.exists(self.config_blocked_sites):
-            with open(self.config_blocked_sites, "w", encoding='utf-8'):
-                pass
-        if not os.path.exists(self.config_blocked_url):
-            with open(self.config_blocked_url, "w", encoding='utf-8'):
-                pass
+        if self.filter_mode == "local":
+            if not os.path.exists(self.config_blocked_sites):
+                with open(self.config_blocked_sites, "w", encoding='utf-8'):
+                    pass
+            if not os.path.exists(self.config_blocked_url):
+                with open(self.config_blocked_url, "w", encoding='utf-8'):
+                    pass
 
         if not self.no_filter:
             self.filter_proc = multiprocessing.Process(
                 target=filter_process,
                 args=(
                     self.queue,
                     self.result_queue,
+                    self.filter_mode,
                     self.config_blocked_sites,
                     self.config_blocked_url
                 )