refactor(prompts): enhance input validation and streamline get_input function

Author: ayanrajpoot10

bugscanx/modules/others/file_toolkit.py

@@ -34,8 +34,8 @@ def write_lines(file_path, lines):
 
 
 def split_file():
-    file_path = get_input("Enter filename", "file")
-    parts = int(get_input("Number of parts", "number"))
+    file_path = get_input("Enter filename", input_type="file", validators="file")
+    parts = int(get_input("Number of parts", validators="number"))
     lines = read_lines(file_path)
     if not lines:
         return
@@ -89,7 +89,7 @@ def merge_files():
 
 
 def clean_file():
-    input_file = get_input("Enter filename", "file")
+    input_file = get_input("Enter filename", input_type="file", validators="file")
     domain_output_file = get_input("Enter domains output filename")
     ip_output_file = get_input("Enter IP output filename")
 
@@ -117,7 +117,7 @@ def clean_file():
 
 
 def remove_duplicates():
-    file_path = get_input("Enter filename", "file")
+    file_path = get_input("Enter filename", input_type="file", validators="file")
     lines = read_lines(file_path)
     if not lines:
         return
@@ -135,7 +135,7 @@ def remove_duplicates():
 
 
 def filter_by_tlds():
-    file_path = get_input("Enter filename", "file")
+    file_path = get_input("Enter filename", input_type="file", validators="file")
     tlds_input = get_input("Enter TLDs ", instruction="(e.g. com, org)")
 
     domains = read_lines(file_path)
@@ -172,7 +172,7 @@ def filter_by_tlds():
 
 
 def filter_by_keywords():
-    file_path = get_input("Enter filename", "file")
+    file_path = get_input("Enter filename", input_type="file", validators="file")
     keywords = [k.strip().lower() for k in get_input("Enter keyword(s)").split(',')]
     output_file = get_input("Enter output filename")
 
@@ -195,7 +195,7 @@ def filter_by_keywords():
 
 
 def cidr_to_ip():
-    cidr_input = get_input("Enter CIDR range")
+    cidr_input = get_input("Enter CIDR range", validators="cidr")
     output_file = get_input("Enter output filename")
 
     try:
@@ -215,7 +215,7 @@ def cidr_to_ip():
 
 
 def domains_to_ip():
-    file_path = get_input("Enter filename", "file")
+    file_path = get_input("Enter filename", input_type="file", validators="cidr")
     output_file = get_input("Enter output filename")
 
     domains = read_lines(file_path)

bugscanx/modules/others/host_info.py

@@ -196,12 +196,12 @@ def scan(self):
 
 
 def main():
-    host = get_input("Enter host")
-    protocol = get_input("Select protocol", "choice", choices=["http", "https"])
+    host = get_input("Enter host", validators="required")
+    protocol = get_input("Select protocol", input_type="choice", choices=["http", "https"])
     available_methods = ["GET", "HEAD", "POST", "PUT", "DELETE", "OPTIONS", "TRACE", "PATCH"]
     method_list = get_input(
         "Select HTTP method(s)",
-        "choice",
+        input_type="choice",
         multiselect=True, 
         choices=available_methods,
         transformer=lambda result: ', '.join(result) if isinstance(result, list) else result

bugscanx/modules/scanners/host_scanner.py

@@ -1,4 +1,4 @@
-from bugscanx.utils.prompts import get_input, get_confirm, is_cidr
+from bugscanx.utils.prompts import get_input, get_confirm
 from bugscanx.utils.cidr import read_cidrs_from_file
 
 
@@ -10,25 +10,23 @@ def get_common_inputs():
     default_filename = "results.txt"
     output = get_input(
         "Enter output filename",
-        default=default_filename,
-        validate_input=False
+        default=default_filename
     )
     threads = get_input(
         "Enter threads",
-        "number",
-        default="50",
-        allow_comma_separated=False
+        validators="number",
+        default="50"
     )
     return output, threads
 
 
 def get_host_input():
-    filename = get_input("Enter filename", "file", mandatory=False)
+    filename = get_input("Enter filename", input_type="file", validators="file", mandatory=False)
     if not filename:
-        cidr = get_input("Enter CIDR range(s)", validators=[is_cidr], mandatory=False)
+        cidr = get_input("Enter CIDR range(s)", validators="cidr", mandatory=False)
         if not cidr:
             cidr_file = get_input(
-                "Enter CIDR file", "file")
+                "Enter CIDR file", input_type="file")
             cidr = read_cidrs_from_file(cidr_file) if cidr_file else None
         return None, cidr
     return filename, None
@@ -39,12 +37,12 @@ def get_input_direct(no302=False):
     if filename is None and cidr is None:
         return None, None, None
 
-    port_list = get_input("Enter port(s)", "number", default="80").split(',')
-    timeout = get_input("Enter timeout (seconds)", "number", default="3")
+    port_list = get_input("Enter port(s)", validators="number", default="80").split(',')
+    timeout = get_input("Enter timeout (seconds)", validators="number", default="3")
     output, threads = get_common_inputs()
     method_list = get_input(
         "Select HTTP method(s)",
-        "choice",
+        input_type="choice",
         multiselect=True, 
         choices=[
             "GET", "HEAD", "POST", "PUT",
@@ -94,8 +92,8 @@ def get_input_proxy():
         "Upgrade: websocket[crlf][crlf]"
     )
     payload = get_input("Enter payload", default=default_payload)
-    port_list = get_input("Enter port(s)", "number", default="80").split(',')
-    output, threads = get_common_inputs(filename or cidr)
+    port_list = get_input("Enter port(s)", validators="number", default="80").split(',')
+    output, threads = get_common_inputs()
 
     if cidr:
         try:
@@ -128,11 +126,11 @@ def get_input_proxy2():
     if filename is None and cidr is None:
         return None, None, None
 
-    port_list = get_input("Enter port(s)", "number", default="80").split(',')
-    output, threads = get_common_inputs(filename or cidr)
+    port_list = get_input("Enter port(s)", validators="number", default="80").split(',')
+    output, threads = get_common_inputs()
     method_list = get_input(
         "Select HTTP method(s)",
-        "choice",
+        input_type="choice",
         multiselect=True, 
         choices=[
             "GET", "HEAD", "POST", "PUT",
@@ -180,7 +178,7 @@ def get_input_ssl():
     if filename is None and cidr is None:
         return None, None, None
 
-    output, threads = get_common_inputs(filename or cidr)
+    output, threads = get_common_inputs()
 
     if cidr:
         try:
@@ -207,8 +205,8 @@ def get_input_ping():
     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)
+    port_list = get_input("Enter port(s)", validators="number", default="443").split(',')
+    output, threads = get_common_inputs()
 
     if cidr:
         try:

bugscanx/modules/scanners/port_scanner.py

@@ -30,7 +30,7 @@ def scan_port(ip, port):
 
 
 def main():    
-    target = get_input("Enter target")
+    target = get_input("Enter target", validators="required")
     try:
         ip = socket.gethostbyname(target)
     except socket.gaierror:
@@ -42,7 +42,7 @@ def main():
 
     scan_type = get_input(
         "Select scan type",
-        "choice",
+        input_type="choice",
         choices=["Common ports", "All ports (1-65535)"]
     )
     ports = COMMON_PORTS if scan_type == "Common ports" else range(1, 65535)
@@ -73,16 +73,5 @@ def main():
                     progress.console.print(f" [green]✓[/] Port {result} is open")
                 progress.advance(task)
 
-    open_ports = sorted(open_ports)
-    
-    console.print("\n[bold green] Scan Results:[/]\n")
-    if open_ports:
-        console.print("[cyan] Open Ports:[/]\n")
-        for port in open_ports:
-            console.print(f" • Port {port}")
-        
-        with open(f"{target}_open_ports.txt", "w") as f:
-            f.write("\n".join(f"Port {port} is open" for port in open_ports))
-        console.print(f"\n[dim] Results saved to {target}_open_ports.txt[/]\n")
-    else:
+    if not open_ports:
         console.print("\n[yellow] No open ports found.[/]\n")

bugscanx/modules/scrapers/iplookup/iplookup.py

@@ -1,7 +1,7 @@
 import os
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
-from bugscanx.utils.prompts import get_input, is_cidr
+from bugscanx.utils.prompts import get_input
 from .sources import get_scrapers
 from .utils import CursorManager, process_input, process_file
 from .logger import IPLookupConsole
@@ -75,11 +75,9 @@ def run(self, ips, output_file, scrapers=None):
 
 def main():
     ips = []
-    input_type = get_input("Select input mode", "choice",
-                          choices=["Manual", "File"])
-
+    input_type = get_input("Select input mode", input_type="choice", choices=["Manual", "File"])
     if input_type == "Manual":
-        ip_input = get_input("Enter IP or CIDR", validators=[is_cidr])
+        ip_input = get_input("Enter IP or CIDR", validators="cidr")
         ips.extend(process_input(ip_input))
         default_output = f"{ip_input}_domains.txt".replace("/", "-")
     else:

bugscanx/modules/scrapers/subfinder/subfinder.py

@@ -1,6 +1,5 @@
 import os
 from concurrent.futures import ThreadPoolExecutor, as_completed
-
 from bugscanx.utils.prompts import get_input
 from .logger import SubFinderConsole
 from .sources import get_sources
@@ -81,15 +80,13 @@ def run(self, domains, output_file, sources):
 def main():
     domains = []
     sources = get_sources()
-    input_type = get_input("Select input mode", "choice",
-                            choices=["Manual", "File"])
-
+    input_type = get_input("Select input mode", input_type="choice", choices=["Manual", "File"])
     if input_type == "Manual":
-        domain_input = get_input("Enter domain(s)")
+        domain_input = get_input("Enter domain(s)", validators="required")
         domains = [d.strip() for d in domain_input.split(',') if DomainValidator.is_valid_domain(d.strip())]
         default_output = f"{domains[0]}_subdomains.txt" if domains else "subdomains.txt"
     else:
-        file_path = get_input("Enter filename", "file")
+        file_path = get_input("Enter filename", input_type="file", validators="file")
         with open(file_path, 'r') as f:
             domains = [d.strip() for d in f if DomainValidator.is_valid_domain(d.strip())]
         default_output = f"{file_path.rsplit('.', 1)[0]}_subdomains.txt"

bugscanx/utils/prompts.py

@@ -1,13 +1,12 @@
 import os
-
 from InquirerPy import get_style
 from InquirerPy.prompts import (
     ListPrompt as select,
     FilePathPrompt as filepath,
     InputPrompt as text,
     ConfirmPrompt as confirm,
 )
-from .validators import create_validator, required, is_file, is_digit, is_cidr
+from .validators import create_validator, VALIDATORS
 
 
 DEFAULT_STYLE = get_style(
@@ -20,30 +19,6 @@
 )
 
 
-INPUT_VALIDATORS = {
-    "file": [required, is_file],
-    "number": [required, is_digit],
-    "text": [required],
-}
-
-
-def strip_handler(handler, strip_input):
-    def wrapped(params):
-        result = handler(params)
-        if strip_input and isinstance(result, str):
-            return result.strip()
-        return result
-    return wrapped
-
-
-INPUT_HANDLERS = {
-    "choice": lambda params: select(**params).execute(),
-    "file": lambda params: filepath(**params).execute(),
-    "number": lambda params: text(**params).execute(),
-    "text": lambda params: text(**params).execute(),
-}
-
-
 def get_input(
     message,
     input_type="text",
@@ -53,58 +28,47 @@ def get_input(
     multiselect=False,
     transformer=None,
     style=DEFAULT_STYLE,
-    validate_input=True,
     instruction="",
     mandatory=True,
-    allow_comma_separated=True,
-    strip_input=True,
     **kwargs
 ):
-    common_params = {
-        "message": f" {message.strip()}" + ("" if instruction else ":"),
+    def auto_strip(result):
+        return result.strip() if isinstance(result, str) else result
+    
+    params = {
+        "message": f" {message.strip()}" + (":" if not instruction else ""),
         "default": "" if default is None else str(default),
-        "qmark": kwargs.pop("qmark", ""),
-        "amark": kwargs.pop("amark", ""),
+        "qmark": "",
+        "amark": "",
         "style": style,
         "instruction": instruction + (":" if instruction else ""),
         "mandatory": mandatory,
         "transformer": transformer,
     }
 
-    if validators is None and validate_input:
-        validators = INPUT_VALIDATORS.get(input_type, [])
-        
-    if validate_input and validators:
-        if input_type == "number":
-            validators = [required, lambda x: is_digit(x, allow_comma_separated)]
-        common_params["validate"] = create_validator(validators)
+    if validators:
+        if isinstance(validators, str) and validators in VALIDATORS:
+            params["validate"] = create_validator(*VALIDATORS[validators])
+        elif isinstance(validators, (list, tuple)):
+            params["validate"] = create_validator(*validators)
 
-    input_type_params = {
-        "choice": {
+    if input_type == "choice":
+        params.update({
             "choices": choices,
             "multiselect": multiselect,
-            "transformer": transformer,
-            "show_cursor": kwargs.pop("show_cursor", False),
-        },
-        "file": {
-            "only_files": kwargs.pop("only_files", True),
-        },
-    }
+            "show_cursor": kwargs.get("show_cursor", False),
+        })
+        return select(**params).execute()
 
-    common_params.update(input_type_params.get(input_type, {}))
-    common_params.update(kwargs)
+    elif input_type == "file":
+        params["only_files"] = kwargs.get("only_files", True)
+        return auto_strip(filepath(**params).execute())
 
-    handler = INPUT_HANDLERS.get(input_type)
-    
-    return strip_handler(handler, strip_input)(common_params)
+    else:
+        return auto_strip(text(**params).execute())
 
 
-def get_confirm(
-    message,
-    default=True,
-    style=DEFAULT_STYLE,
-    **kwargs
-):
+def get_confirm(message, default=True, style=DEFAULT_STYLE, **kwargs):
     return confirm(
         message=message,
         default=default,