Feature/ip source

README.md

@@ -38,6 +38,7 @@
 | Web interface monitoring                     | ✅        |
 | Lightweight Docker image                     | ✅        |
 | Proxy chaining (multi-proxy forwarding)      | ✅        |
+| IP whitelist with subnet support             | ✅        |
 
 ## 📦 **Installation**
 
@@ -71,8 +72,6 @@ If you encounter any problems, or if you want to use the program in a particular
 
 - Support content analysis
 - Caching of latest and most searched pages
-- Adding ACL
-- Proxy authentication
 
 ## 🏎️ **Benchmark**
 

config.ini.example

@@ -21,6 +21,7 @@ blocked_url = config/blocked_url.txt
 [Options]
 shortcuts = config/shortcuts.txt
 custom_header = config/custom_header.json
+authorized_ips = config/authorized_ips.txt
 
 [Security]
 ssl_inspect = false

config/authorized_ips.example.txt

@@ -0,0 +1,3 @@
+0.0.0.0/0
+127.0.0.1/8
+10.0.0.1/32

pyproxy.py

@@ -23,6 +23,7 @@ def main():
     html_403 = get_config_value(args, config, 'html_403', 'Files', "assets/403.html")
     shortcuts = get_config_value(args, config, 'shortcuts', 'Options', "config/shortcuts.txt")
     custom_header = get_config_value(args, config, 'custom_header', 'Options', "config/custom_header.json")
+    authorized_ips = get_config_value(args, config, 'authorized_ips', 'Options', "config/authorized_ips.txt")
     flask_port = get_config_value(args, config, 'flask_port', 'Monitoring', 5000)
     flask_pass = get_config_value(args, config, 'flask_pass', 'Monitoring', "password")
     proxy_enable = get_config_value(args, config, 'proxy_enable', 'Proxy', False)
@@ -63,6 +64,7 @@ def main():
         html_403=html_403,
         shortcuts=shortcuts,
         custom_header=custom_header,
+        authorized_ips=authorized_ips,
         proxy_enable=proxy_enable,
         proxy_host=proxy_host,
         proxy_port=proxy_port

pyproxy/server.py

@@ -13,6 +13,7 @@
 import multiprocessing
 import os
 import time
+import ipaddress
 
 from pyproxy.utils.version import __slim__
 from pyproxy.utils.logger import configure_file_logger, configure_console_logger
@@ -43,7 +44,8 @@ class ProxyServer:
 
     def __init__(self, host, port, debug, logger_config, filter_config,
                  html_403, ssl_config, shortcuts, custom_header,
-                 flask_port, flask_pass, proxy_enable, proxy_host, proxy_port):
+                 flask_port, flask_pass, proxy_enable, proxy_host, proxy_port,
+                 authorized_ips):
         """
         Initialize the ProxyServer with configuration parameters.
         """
@@ -61,9 +63,13 @@ def __init__(self, host, port, debug, logger_config, filter_config,
         self.flask_pass = flask_pass
 
         # Proxy
-        self.proxy_enable=proxy_enable
-        self.proxy_host=proxy_host
-        self.proxy_port=proxy_port
+        self.proxy_enable = proxy_enable
+        self.proxy_host = proxy_host
+        self.proxy_port = proxy_port
+
+        # Authorized IPS
+        self.authorized_ips = authorized_ips
+        self.allowed_subnets = None
 
         # Process communication queues
         self.filter_proc = None
@@ -163,6 +169,26 @@ def _clean_inspection_folder(self):
                 except (FileNotFoundError, PermissionError, OSError) as e:
                     self.console_logger.debug("Error deleting %s: %s", file_path, e)
 
+    def _load_authorized_ips(self):
+        """
+        Load authorized IPs/subnets from the file.
+        """
+        self.allowed_subnets = None
+
+        if self.authorized_ips and os.path.isfile(self.authorized_ips):
+            with open(self.authorized_ips, "r", encoding="utf-8") as f:
+                lines = [line.strip() for line in f if line.strip()]
+            try:
+                self.allowed_subnets = [ipaddress.ip_network(line, strict=False) for line in lines]
+                self.console_logger.debug(
+                    "[*] Loaded %d authorized IPs/subnets",
+                    len(self.allowed_subnets)
+                )
+            except ValueError as e:
+                self.console_logger.error("[*] Invalid IP/subnet in %s: %s", self.authorized_ips, e)
+                self.allowed_subnets = None
+
+    # pylint: disable=R0912
     def start(self):
         """
         Start the proxy server and listen for incoming client connections.
@@ -201,6 +227,7 @@ def start(self):
                         pass
 
         self._initialize_processes()
+        self._load_authorized_ips()
 
         if not __slim__:
             flask_thread = threading.Thread(
@@ -219,6 +246,15 @@ def start(self):
         try:
             while True:
                 client_socket, addr = server.accept()
+                client_ip, client_port = addr
+
+                if self.allowed_subnets:
+                    ip_obj = ipaddress.ip_address(client_ip)
+                    if not any(ip_obj in net for net in self.allowed_subnets):
+                        self.console_logger.debug("Unauthorized IP blocked: %s", client_ip)
+                        client_socket.close()
+                        continue
+
                 self.console_logger.debug("Connection from %s", addr)
                 client = ProxyHandlers(
                     html_403=self.html_403,

pyproxy/utils/args.py

@@ -40,6 +40,7 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--blocked-url", type=str, help="Path to the text file containing the list of URLs to block")
     parser.add_argument("--shortcuts", type=str, help="Path to the text file containing the list of shortcuts")
     parser.add_argument("--custom-header", type=str, help="Path to the json file containing the list of custom headers")
+    parser.add_argument("--authorized-ips", type=str, help="Path to the txt file containing the list of authorized ips")
     parser.add_argument("--no-logging-access", action="store_true", help="Disable access logging")
     parser.add_argument("--no-logging-block", action="store_true", help="Disable block logging")
     parser.add_argument("--ssl-inspect", action="store_true", help="Enable SSL inspection")