refactor(scanners): replace host_list with input_file in scanner classes for generator based file handling and task generation for hosts

Author: ayanrajpoot10

bugscanx/modules/scanners_pro/concurrency/multithread.py

@@ -21,12 +21,9 @@ def __init__(self, threads=50):
 
     def _add_task(self, task):
         self._queue.put(task)
-        if not hasattr(self, '_total_preset'):
-            self._total += 1
 
     def set_total(self, total):
         self._total = total
-        self._total_preset = True
 
     def start(self):
         print()

bugscanx/modules/scanners_pro/host_scanner_pro.py

@@ -4,15 +4,6 @@
 from bugscanx.utils.common import get_input, get_confirm, is_cidr
 
 
-def read_hosts(filename=None, cidr=None):
-    if filename:
-        with open(filename) as file:
-            return [line.strip() for line in file]
-    elif cidr:
-        return []
-    return []
-
-
 def get_cidr_ranges_from_input(cidr_input):
     return [c.strip() for c in cidr_input.split(',')]
 
@@ -65,7 +56,7 @@ def get_input_direct(no302=False):
 
     if cidr:
         cidr_ranges = get_cidr_ranges_from_input(cidr)
-        from .scanners.direct import CIDRDirectScanner
+        from .scanners.direct import CIDRDirectScanner        
         scanner = CIDRDirectScanner(
             method_list=method_list,
             cidr_ranges=cidr_ranges,
@@ -76,7 +67,7 @@ def get_input_direct(no302=False):
         from .scanners.direct import HostDirectScanner
         scanner = HostDirectScanner(
             method_list=method_list,
-            host_list=read_hosts(filename, cidr),
+            input_file=filename,
             port_list=port_list,
             no302=no302
         )
@@ -112,7 +103,7 @@ def get_input_proxy():
     else:
         from .scanners.proxy_check import HostProxyScanner
         scanner = HostProxyScanner(
-            host_list=read_hosts(filename, cidr),
+            input_file=filename,
             port_list=port_list,
             target=target_url,
             payload=payload,
@@ -161,7 +152,7 @@ def get_input_proxy2():
         from .scanners.proxy_request import HostProxy2Scanner
         scanner = HostProxy2Scanner(
             method_list=method_list,
-            host_list=read_hosts(filename, cidr),
+            input_file=filename,
             port_list=port_list,
         ).set_proxy(proxy, proxy_username, proxy_password)
 
@@ -184,14 +175,16 @@ def get_input_ssl():
     else:
         from .scanners.ssl import HostSSLScanner
         scanner = HostSSLScanner(
-            host_list=read_hosts(filename, cidr),
+            input_file=filename,
         )
 
     return scanner, output, threads
 
 
 def get_input_ping():
     filename, cidr = get_host_input()
+    if filename is None and cidr is None:
+        return None, None, None
 
     port_list = get_input("Enter port(s)", "number", default="443").split(',')
     output, threads = get_common_inputs(filename or cidr)
@@ -206,7 +199,7 @@ def get_input_ping():
     else:
         from .scanners.ping import HostPingScanner
         scanner = HostPingScanner(
-            host_list=read_hosts(filename, cidr),
+            input_file=filename,
             port_list=port_list,
         )
 

bugscanx/modules/scanners_pro/scanners/base.py

@@ -16,18 +16,6 @@ def get_url(self, host, port):
         protocol = 'https' if port == '443' else 'http'
         return f'{protocol}://{self.convert_host_port(host, port)}'
 
-    def filter_list(self, data):
-        if self.is_cidr_input:
-            return data
-            
-        filtered_data = []
-        for item in data:
-            item = str(item).strip()
-            if item.startswith(('#', '*')) or not item:
-                continue
-            filtered_data.append(item)
-        return list(set(filtered_data))
-
     def generate_cidr_hosts(self, cidr_ranges):
         for cidr in cidr_ranges:
             try:
@@ -53,3 +41,27 @@ def set_cidr_total(self, cidr_ranges):
             port_multiplier = len(getattr(self, 'port_list', [1]))
             method_multiplier = len(getattr(self, 'method_list', [1]))
             self.set_total(total_hosts * port_multiplier * method_multiplier)
+
+    def read_lines_count(self, filepath):
+        line_count = 0
+        with open(filepath, 'rb') as f:
+            while chunk := f.read(8192):
+                line_count += chunk.count(b'\n')
+        return line_count
+
+    def set_host_total(self, host_filepath):
+        if host_filepath:
+            total_hosts = self.read_lines_count(host_filepath)
+            port_multiplier = len(getattr(self, 'port_list', [1]))
+            method_multiplier = len(getattr(self, 'method_list', [1]))
+            self.set_total(total_hosts * port_multiplier * method_multiplier)
+
+    def generate_hosts_from_file(self, filepath):
+        try:
+            with open(filepath, 'r', encoding='utf-8') as file:
+                for line in file:
+                    host = line.strip()
+                    if host and not host.startswith(('#', '*')):
+                        yield host
+        except (FileNotFoundError, IOError, UnicodeDecodeError):
+            return

bugscanx/modules/scanners_pro/scanners/direct.py

@@ -92,7 +92,7 @@ class HostDirectScanner(DirectScannerBase):
     def __init__(
         self,
         method_list=None,
-        host_list=None,
+        input_file=None,
         port_list=None,
         no302=False,
         threads=50,
@@ -106,7 +106,10 @@ def __init__(
             is_cidr_input=False,
             **kwargs
         )
-        self.host_list = host_list or []
+        self.input_file = input_file
+        
+        if self.input_file:
+            self.set_host_total(self.input_file)
 
     def log_info(self, **kwargs):
         server = kwargs.get('server', '')
@@ -119,15 +122,13 @@ def log_info(self, **kwargs):
             self.logger.colorize(f"{{server:<15}}", "MAGENTA"),
             self.logger.colorize(f"{{port:<4}}", "ORANGE"),
             self.logger.colorize(f"{{ip:<16}}", "BLUE"),
-            self.logger.colorize(f"{{host}}", "LGRAY")
-        ]
-
+            self.logger.colorize(f"{{host}}", "LGRAY")        ]
         self.logger.log('  '.join(messages).format(**kwargs))
 
     def generate_tasks(self):
-        for method in self.filter_list(self.method_list):
-            for host in self.filter_list(self.host_list):
-                for port in self.filter_list(self.port_list):
+        for method in self.method_list:
+            for host in self.generate_hosts_from_file(self.input_file):
+                for port in self.port_list:
                     yield {
                         'method': method.upper(),
                         'host': host,
@@ -180,12 +181,11 @@ def log_info(self, **kwargs):
             self.logger.colorize(f"{{host}}", "LGRAY")
         ]
 
-        self.logger.log('  '.join(messages).format(**kwargs))
-
+        self.logger.log('  '.join(messages).format(**kwargs))    
     def generate_tasks(self):
-        for method in self.filter_list(self.method_list):
+        for method in self.method_list:
             for host in self.generate_cidr_hosts(self.cidr_ranges):
-                for port in self.filter_list(self.port_list):
+                for port in self.port_list:
                     yield {
                         'method': method.upper(),
                         'host': host,

bugscanx/modules/scanners_pro/scanners/ping.py

@@ -49,13 +49,16 @@ def complete(self):
 class HostPingScanner(PingScannerBase):    
     def __init__(
         self,
-        host_list=None,
+        input_file=None,
         port_list=None,
         threads=50,
         **kwargs
     ):
         super().__init__(port_list=port_list, threads=threads, is_cidr_input=False, **kwargs)
-        self.host_list = host_list or []
+        self.input_file = input_file
+
+        if self.input_file:
+            self.set_host_total(self.input_file)
 
     def init(self):
         self.log_info(port='Port', ip='IP', host='Host')
@@ -75,17 +78,16 @@ def _handle_success(self, data):
         data['ip'] = ip
         self.success(data)
         self.log_info(**data)
-
+        
     def generate_tasks(self):
-        for host in self.filter_list(self.host_list):
-            for port in self.filter_list(self.port_list):
+        for host in self.generate_hosts_from_file(self.input_file):
+            for port in self.port_list:
                 yield {
                     'host': host,
                     'port': port,
                 }
 
 
-
 class CIDRPingScanner(PingScannerBase):    
     def __init__(
         self,
@@ -114,7 +116,7 @@ def _handle_success(self, data):
 
     def generate_tasks(self):
         for host in self.generate_cidr_hosts(self.cidr_ranges):
-            for port in self.filter_list(self.port_list):
+            for port in self.port_list:
                 yield {
                     'host': host,
                     'port': port,

bugscanx/modules/scanners_pro/scanners/proxy_check.py

@@ -82,7 +82,7 @@ def log_info(self, proxy_host_port, status_code, response_lines=None):
 class HostProxyScanner(ProxyScannerBase):
     def __init__(
         self,
-        host_list=None,
+        input_file=None,
         port_list=None,
         target='',
         payload='',
@@ -97,11 +97,14 @@ def __init__(
             is_cidr_input=False,
             **kwargs
         )
-        self.host_list = host_list or []
+        self.input_file = input_file
+
+        if self.input_file:
+            self.set_host_total(self.input_file)
 
     def generate_tasks(self):
-        for proxy_host in self.filter_list(self.host_list):
-            for port in self.filter_list(self.port_list):
+        for proxy_host in self.generate_hosts_from_file(self.input_file):
+            for port in self.port_list:
                 yield {
                     'proxy_host': proxy_host,
                     'port': port,
@@ -138,7 +141,7 @@ def __init__(
 
     def generate_tasks(self):
         for proxy_host in self.generate_cidr_hosts(self.cidr_ranges):
-            for port in self.filter_list(self.port_list):
+            for port in self.port_list:
                 yield {
                     'proxy_host': proxy_host,
                     'port': port,

bugscanx/modules/scanners_pro/scanners/proxy_request.py

@@ -67,7 +67,7 @@ class HostProxy2Scanner(Proxy2ScannerBase):
     def __init__(
         self,
         method_list=None,
-        host_list=None,
+        input_file=None,
         port_list=None,
         no302=False,
         threads=50,
@@ -81,7 +81,10 @@ def __init__(
             is_cidr_input=False,
             **kwargs
         )
-        self.host_list = host_list or []
+        self.input_file = input_file
+
+        if self.input_file:
+            self.set_host_total(self.input_file)
 
     def log_info(self, **kwargs):
         server = kwargs.get('server', '')
@@ -100,9 +103,9 @@ def log_info(self, **kwargs):
         self.logger.log('  '.join(messages).format(**kwargs))
 
     def generate_tasks(self):
-        for method in self.filter_list(self.method_list):
-            for host in self.filter_list(self.host_list):
-                for port in self.filter_list(self.port_list):
+        for method in self.method_list:
+            for host in self.generate_hosts_from_file(self.input_file):
+                for port in self.port_list:
                     yield {
                         'method': method.upper(),
                         'host': host,
@@ -158,9 +161,9 @@ def log_info(self, **kwargs):
         self.logger.log('  '.join(messages).format(**kwargs))
 
     def generate_tasks(self):
-        for method in self.filter_list(self.method_list):
+        for method in self.method_list:
             for host in self.generate_cidr_hosts(self.cidr_ranges):
-                for port in self.filter_list(self.port_list):
+                for port in self.port_list:
                     yield {
                         'method': method.upper(),
                         'host': host,

bugscanx/modules/scanners_pro/scanners/ssl.py

@@ -52,12 +52,15 @@ def complete(self):
 class HostSSLScanner(SSLScannerBase):
     def __init__(
         self,
-        host_list=None,
+        input_file=None,
         threads=50,
         **kwargs
     ):
         super().__init__(threads=threads, is_cidr_input=False, **kwargs)
-        self.host_list = host_list or []
+        self.input_file = input_file
+
+        if self.input_file:
+            self.set_host_total(self.input_file)
 
     def log_info(self, **kwargs):
         messages = [
@@ -68,7 +71,7 @@ def log_info(self, **kwargs):
         self.logger.log('  '.join(messages).format(**kwargs))
 
     def generate_tasks(self):
-        for host in self.filter_list(self.host_list):
+        for host in self.generate_hosts_from_file(self.input_file):
             yield {
                 'host': host,
             }