refactor(subfinder): rewrite module using OOP principles and cleaner architecture

- Improved structure by applying object-oriented design patterns
- Enhanced readability and maintainability through modular and clean code practices
- Removed redundant logic and simplified control flow

Author: ayanrajpoot10

bugscanx/modules/scrapers/subfinder/logger.py

@@ -1,50 +1,36 @@
-from threading import RLock
 from rich.console import Console
 
-console = Console()
 
-class Logger:
+class SubFinderConsole(Console):
     def __init__(self):
-        self._lock = RLock()
+        super().__init__()
+        self.total_subdomains = 0
+        self.domain_stats = {}
+        self.disable_cursor()
 
-    def clear_line(self):
-        with self._lock:
-            print("\033[2K\r", end='', flush=True)
+    def disable_cursor(self):
+        print('\033[?25l', end='', flush=True)
 
-    def replace(self, message):
-        with self._lock:
-            print(f"{message}", end='', flush=True)
+    def enable_cursor(self):
+        print('\033[?25h', end='', flush=True)
 
-class SubFinderConsole:
-    def __init__(self):
-        self.total_subdomains = 0
-        self.domain_stats = {}
-        self.logger = Logger()
-    
-    def start_domain_scan(self, domain):
-        self.logger.clear_line()
-        console.print(f"[cyan] Processing: {domain}[/cyan]")
+    def print_domain_start(self, domain):
+        self.print(f"[cyan]Processing: {domain}[/cyan]")
 
     def update_domain_stats(self, domain, count):
         self.domain_stats[domain] = count
         self.total_subdomains += count
 
-    def print_domain_complete(self, domain, subdomains_count):
-        self.logger.clear_line()
-        console.print(f"[green] {domain}: {subdomains_count} subdomains found[/green]")
+    def print_domain_complete(self, domain, count):
+        self.print(f"[green]{domain}: {count} subdomains found[/green]")
 
     def print_final_summary(self, output_file):
-        console.print(f"\n[green] Total: [bold]{self.total_subdomains}[/bold] subdomains found[/green]")
-        console.print(f"[green] Results saved to {output_file}[/green]")
+        self.print(f"\n[green]Total: [bold]{self.total_subdomains}[/bold] subdomains found")
+        self.print(f"[green]Results saved to {output_file}[/green]")
+        self.enable_cursor()
 
-    def show_progress(self, current, total):
-        progress_message = f" progress: [{current}/{total}]\r"
-        self.logger.replace(progress_message)
-    
-    def print(self, message):
-        self.logger.clear_line()
-        console.print(message)
+    def print_progress(self, current, total):
+        self.print(f"Progress: {current} / {total}", end="\r")
 
     def print_error(self, message):
-        self.logger.clear_line()
-        console.print(f"[red] {message}[/red]")
+        self.print(f"[red]{message}[/red]")

bugscanx/modules/scrapers/subfinder/sources.py

@@ -1,21 +1,23 @@
+from abc import ABC, abstractmethod
 from bs4 import BeautifulSoup
+from .utils import RequestHandler
 
-from .utils import make_request
-
-class SubdomainSource:
+class SubdomainSource(RequestHandler, ABC):
     def __init__(self, name):
+        super().__init__()
         self.name = name
         self.subdomains = set()
-    
-    def fetch(self, domain, session=None):
-        raise NotImplementedError
+
+    @abstractmethod
+    def fetch(self, domain):
+        pass
 
 class CrtshSource(SubdomainSource):
     def __init__(self):
         super().__init__("Crt.sh")
-    
-    def fetch(self, domain, session=None):
-        response = make_request(f"https://crt.sh/?q=%25.{domain}&output=json", session)
+
+    def fetch(self, domain):
+        response = self.get(f"https://crt.sh/?q=%25.{domain}&output=json")
         if response and response.headers.get('Content-Type') == 'application/json':
             for entry in response.json():
                 self.subdomains.update(entry['name_value'].splitlines())
@@ -24,19 +26,21 @@ def fetch(self, domain, session=None):
 class HackertargetSource(SubdomainSource):
     def __init__(self):
         super().__init__("Hackertarget")
-    
-    def fetch(self, domain, session=None):
-        response = make_request(f"https://api.hackertarget.com/hostsearch/?q={domain}", session)
+
+    def fetch(self, domain):
+        response = self.get(f"https://api.hackertarget.com/hostsearch/?q={domain}")
         if response and 'text' in response.headers.get('Content-Type', ''):
-            self.subdomains.update([line.split(",")[0] for line in response.text.splitlines()])
+            self.subdomains.update(
+                [line.split(",")[0] for line in response.text.splitlines()]
+            )
         return self.subdomains
 
 class RapidDnsSource(SubdomainSource):
     def __init__(self):
         super().__init__("RapidDNS")
-    
-    def fetch(self, domain, session=None):
-        response = make_request(f"https://rapiddns.io/subdomain/{domain}?full=1", session)
+
+    def fetch(self, domain):
+        response = self.get(f"https://rapiddns.io/subdomain/{domain}?full=1")
         if response:
             soup = BeautifulSoup(response.text, 'html.parser')
             for link in soup.find_all('td'):
@@ -48,19 +52,19 @@ def fetch(self, domain, session=None):
 class AnubisDbSource(SubdomainSource):
     def __init__(self):
         super().__init__("AnubisDB")
-    
-    def fetch(self, domain, session=None):
-        response = make_request(f"https://jldc.me/anubis/subdomains/{domain}", session)
+
+    def fetch(self, domain):
+        response = self.get(f"https://jldc.me/anubis/subdomains/{domain}")
         if response:
             self.subdomains.update(response.json())
         return self.subdomains
 
 class AlienVaultSource(SubdomainSource):
     def __init__(self):
         super().__init__("AlienVault")
-    
-    def fetch(self, domain, session=None):
-        response = make_request(f"https://otx.alienvault.com/api/v1/indicators/domain/{domain}/passive_dns", session)
+
+    def fetch(self, domain):
+        response = self.get(f"https://otx.alienvault.com/api/v1/indicators/domain/{domain}/passive_dns")
         if response:
             for entry in response.json().get("passive_dns", []):
                 hostname = entry.get("hostname")
@@ -71,25 +75,15 @@ def fetch(self, domain, session=None):
 class CertSpotterSource(SubdomainSource):
     def __init__(self):
         super().__init__("CertSpotter")
-    
-    def fetch(self, domain, session=None):
-        response = make_request(f"https://api.certspotter.com/v1/issuances?domain={domain}&include_subdomains=true&expand=dns_names", session)
+
+    def fetch(self, domain):
+        response = self.get(f"https://api.certspotter.com/v1/issuances?domain={domain}&include_subdomains=true&expand=dns_names")
         if response:
             for cert in response.json():
                 self.subdomains.update(cert.get('dns_names', []))
         return self.subdomains
 
-def get_all_sources():
-    return [
-        CrtshSource(),
-        HackertargetSource(),
-        RapidDnsSource(),
-        AnubisDbSource(),
-        AlienVaultSource(),
-        CertSpotterSource(),
-    ]
-
-def get_bulk_sources():
+def get_sources():
     return [
         CrtshSource(),
         HackertargetSource(),

bugscanx/modules/scrapers/subfinder/subfinder.py

@@ -1,55 +1,51 @@
 import os
-import requests
-import threading
-from rich import print
 from concurrent.futures import ThreadPoolExecutor, as_completed
+
 from bugscanx.utils.common import get_input
 from .logger import SubFinderConsole
-from .sources import get_all_sources, get_bulk_sources
+from .sources import get_sources
 from .utils import is_valid_domain, filter_valid_subdomains
 
 class SubFinder:
     def __init__(self):
         self.console = SubFinderConsole()
-    
-    def process_domain(self, domain, output_file, sources, total, completed_counter):
+        self.completed = 0
+
+    def _fetch_from_source(self, source, domain):
+        try:
+            found = source.fetch(domain)
+            return filter_valid_subdomains(found, domain)
+        except Exception:
+            return set()
+
+    def _save_subdomains(self, subdomains, output_file):
+        if subdomains:
+            with open(output_file, "a", encoding="utf-8") as f:
+                f.write("\n".join(sorted(subdomains)) + "\n")
+
+    def process_domain(self, domain, output_file, sources, total):
         if not is_valid_domain(domain):
-            with completed_counter.get_lock():
-                completed_counter.value += 1
+            self.completed += 1
             return set()
 
-        self.console.start_domain_scan(domain)
-        self.console.show_progress(completed_counter.value, total)
+        self.console.print_domain_start(domain)
+        self.console.print_progress(self.completed, total)
 
-        with requests.Session() as session:
-            results = []
-            with ThreadPoolExecutor(max_workers=6) as executor:
-                future_to_source = {
-                    executor.submit(source.fetch, domain, session): source.name
-                    for source in sources
-                }
-                
-                for future in as_completed(future_to_source):
-                    try:
-                        found = future.result()
-                        filtered = filter_valid_subdomains(found, domain)
-                        results.append(filtered)
-                    except Exception:
-                        results.append(set())
-            
-            subdomains = set().union(*results) if results else set()
+        with ThreadPoolExecutor(max_workers=6) as executor:
+            futures = [
+                executor.submit(self._fetch_from_source, source, domain)
+                for source in sources
+            ]
+            results = [f.result() for f in as_completed(futures)]
+
+        subdomains = set().union(*results) if results else set()
 
         self.console.update_domain_stats(domain, len(subdomains))
         self.console.print_domain_complete(domain, len(subdomains))
+        self._save_subdomains(subdomains, output_file)
 
-        if subdomains:
-            with open(output_file, "a", encoding="utf-8") as f:
-                f.write("\n".join(sorted(subdomains)) + "\n")
-
-        with completed_counter.get_lock():
-            completed_counter.value += 1
-            self.console.show_progress(completed_counter.value, total)
-
+        self.completed += 1
+        self.console.print_progress(self.completed, total)
         return subdomains
 
     def run(self, domains, output_file, sources):
@@ -58,55 +54,42 @@ def run(self, domains, output_file, sources):
             return
 
         os.makedirs(os.path.dirname(output_file) or '.', exist_ok=True)
-        completed_counter = threading.Value('i', 0)
+        self.completed = 0
         all_subdomains = set()
-        
+        total = len(domains)
+
         with ThreadPoolExecutor(max_workers=3) as executor:
-            future_to_domain = {
-                executor.submit(
-                    self.process_domain, 
-                    domain, 
-                    output_file, 
-                    sources, 
-                    len(domains), 
-                    completed_counter
-                ): domain for domain in domains
-            }
-            
-            for future in as_completed(future_to_domain):
-                domain = future_to_domain[future]
+            futures = [
+                executor.submit(self.process_domain, domain, output_file, sources, total)
+                for domain in domains
+            ]
+            for future in as_completed(futures):
                 try:
                     result = future.result()
-                    if result:
-                        all_subdomains.update(result)
+                    all_subdomains.update(result)
                 except Exception as e:
-                    self.console.print(f"Error processing {domain}: {str(e)}")
+                    self.console.print(f"Error processing domain: {str(e)}")
 
         self.console.print_final_summary(output_file)
         return all_subdomains
 
+
 def main():
     domains = []
-    input_type = get_input("Select input mode", "choice", 
-                        choices=["Manual", "File"])
-    
+    sources = get_sources()
+    input_type = get_input("Select input mode", "choice",
+                           choices=["Manual", "File"])
+
     if input_type == "Manual":
         domain_input = get_input("Enter domain(s)")
         domains = [d.strip() for d in domain_input.split(',') if is_valid_domain(d.strip())]
-        sources = get_all_sources()
         default_output = f"{domains[0]}_subdomains.txt"
-
     else:
         file_path = get_input("Enter filename", "file")
         with open(file_path, 'r') as f:
             domains = [d.strip() for d in f if is_valid_domain(d.strip())]
-        sources = get_bulk_sources()
         default_output = f"{file_path.rsplit('.', 1)[0]}_subdomains.txt"
 
-    if not domains:
-        print("[bold red] No valid domains provided")
-        return
-
     output_file = get_input("Enter output filename", default=default_output)
     subfinder = SubFinder()
     subfinder.run(domains, output_file, sources)