diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 1072587..e6c525b 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -8,13 +8,9 @@ and [issues](https://github.com/vil/H4X-Tools/issues) page to see if there is an ## Submitting changes -After you have forked the project and done your changes. Open a pull request and tell what you have changed, improved +After you have forked the project and done your changes. Open up a pull request and tell what you have changed, improved or added. -Also avoid using AI, human written code is always better. - Don't also write **random** commit messages, simply tell shortly what you have done. - -------------------------- -Thank you for understanding. \ No newline at end of file +Thank you. diff --git a/h4xtools.py b/h4xtools.py index b12e442..e6d897d 100755 --- a/h4xtools.py +++ b/h4xtools.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -24,7 +24,7 @@ from helper import handles, printer -VERSION = "0.3.3" +VERSION = "0.3.4" def internet_check() -> None: @@ -160,9 +160,7 @@ def main() -> None: user_input = printer.user_input("Tool to execute : \t") if user_input.lower() in {"quit", "exit", "q", "kill"}: - """ - Kills the program. - """ + # Kill the program. printer.warning("Quitting... Goodbye!") print(Style.RESET_ALL) time.sleep(0.5) @@ -184,13 +182,14 @@ def main() -> None: if __name__ == "__main__": - try: - main() - except ValueError: - printer.error("Invalid value inputted..!") - main() # re-run - except KeyboardInterrupt: - print("\n") - printer.warning("Quitting... Goodbye!") - print(Style.RESET_ALL) - exit(0) + while True: + try: + main() + break + except ValueError: + printer.error("Invalid value inputted..!") + except KeyboardInterrupt: + print("\n") + printer.warning("Quitting... Goodbye!") + print(Style.RESET_ALL) + exit(0) diff --git a/helper/printer.py b/helper/printer.py index 734649a..d8c61b9 100644 --- a/helper/printer.py +++ b/helper/printer.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -55,8 +55,6 @@ def warning(message, *args, **kwargs) -> None: def debug(message, *args, **kwargs) -> None: if "--debug" in sys.argv: print_colored(message, Fore.LIGHTMAGENTA_EX, "[>]", *args, **kwargs) - else: - pass def noprefix(message, *args, **kwargs) -> None: diff --git a/helper/randomuser.py b/helper/randomuser.py index a52b036..f739f29 100644 --- a/helper/randomuser.py +++ b/helper/randomuser.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by diff --git a/helper/timer.py b/helper/timer.py index a1a6047..047ef5e 100644 --- a/helper/timer.py +++ b/helper/timer.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -15,6 +15,7 @@ along with this program. If not, see . """ +import functools import time from types import FunctionType @@ -30,6 +31,7 @@ def timer(require_input: bool) -> FunctionType: """ def decorator(func): + @functools.wraps(func) def wrapper(*args, **kwargs) -> str: start_time = time.time() # Start timing result = func(*args, **kwargs) # Execute the wrapped function diff --git a/helper/url_helper.py b/helper/url_helper.py index 0499665..0ca7508 100644 --- a/helper/url_helper.py +++ b/helper/url_helper.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -22,12 +22,13 @@ from helper import printer -def read_local_content(path) -> str: +def read_local_content(path) -> str | dict | None: """ Reads file content from a local file. :param path: path to the file - :return: Content of the file as a string or None if an error occurs. + :return: Parsed dict for JSON files, raw string for text files, + or None if an error occurs. """ try: with open(resource_path(path), "r") as file: @@ -38,7 +39,7 @@ def read_local_content(path) -> str: return content except Exception as e: printer.error(f"An error occurred: {str(e)}") - return "None" + return None # I hate pyinstaller. diff --git a/requirements.txt b/requirements.txt index d590812..a60493e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,11 +5,11 @@ beautifulsoup4 urllib3 pyinstaller whoisdomain -requests git+https://github.com/diezo/Ensta.git snscrape faker holehe +ignorant +httpx aiohttp -asyncio psutil diff --git a/utils/bluetooth_scanner.py b/utils/bluetooth_scanner.py index 76068c2..79269e1 100644 --- a/utils/bluetooth_scanner.py +++ b/utils/bluetooth_scanner.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -65,6 +65,10 @@ def scan_linux(duration: int) -> str: bufsize=1, ) + if process.stdin is None: + printer.error("Failed to open bluetoothctl stdin.") + return output + process.stdin.write("scan on\n") process.stdin.flush() @@ -105,20 +109,23 @@ def parse_output(output: str, platform: str) -> None: for line in clean_output.splitlines(): printer.debug(line) if "[NEW] Device" in line: - parts = line.split(" ", 4) + parts = line.split() printer.debug(parts) # parts[0] = [NEW] # parts[1] = Device - # parts[2] = MAC - # parts[3+] = device name + # parts[2] = MAC address + # parts[3:] = device name (may contain spaces) + if len(parts) < 3: + continue mac = parts[2] - name = parts[4].strip() if len(parts) > 4 else "No Name" + name = " ".join(parts[3:]).strip() if len(parts) > 3 else "No Name" devices.append({"mac": mac, "name": name, "raw": line}) printer.info("Nearby devices :") for device in devices: printer.success(f"Device: {device['name']} ({device['mac']})") - printer.success(f"RAW: {device['raw']}", "\n") + printer.success(f"RAW: {device['raw']}") + print() case _: printer.error("idk how u got here.") diff --git a/utils/dirbuster.py b/utils/dirbuster.py index 562b593..5c04c03 100644 --- a/utils/dirbuster.py +++ b/utils/dirbuster.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -18,89 +18,99 @@ import asyncio import aiohttp -import requests from colorama import Style from helper import printer, randomuser, timer, url_helper -url_set = set() -target_domain: str | None = None - @timer.timer(require_input=True) def bust(domain: str) -> None: """ - Scans the given url for valid paths + Scans the given url for valid paths. - param domain: url to scan + :param domain: url to scan """ - target_domain = domain - printer.info( - f"Scanning for valid URLs for {Style.BRIGHT}{target_domain}{Style.RESET_ALL}..." + f"Scanning for valid URLs for {Style.BRIGHT}{domain}{Style.RESET_ALL}..." ) printer.warning("This may take a while...") - scan_urls() + valid_urls = scan_urls(domain) printer.success( - f"Scan Completed..! There were {Style.BRIGHT}{len(url_set)}{Style.RESET_ALL} valid URLs!" + f"Scan Completed..! There were {Style.BRIGHT}{len(valid_urls)}{Style.RESET_ALL} valid URLs!" ) def get_wordlist() -> set[str] | None: """ - Reads the wordlist from the url and returns a list of names + Reads the wordlist from the local resources and returns a set of paths. - :return: list of names + :return: set of path strings, or None if an error occurs. """ - try: - content = url_helper.read_local_content("resources/wordlist.txt") - return {line.strip() for line in content.splitlines() if line.strip()} - except requests.exceptions.ConnectionError: + content = url_helper.read_local_content("resources/wordlist.txt") + if not isinstance(content, str): return None + return {line.strip() for line in content.splitlines() if line.strip()} -async def fetch_url(session: aiohttp.ClientSession, path: str) -> None: +async def fetch_url( + session: aiohttp.ClientSession, domain: str, path: str, url_set: set +) -> None: """ - Fetches the url and checks if it is valid + Fetches a URL and records it if it returns HTTP 200. :param session: aiohttp session + :param domain: target domain :param path: path to check + :param url_set: set to collect valid URLs into """ - url = f"https://{target_domain}/{path}" - headers = {"User-Agent": f"{randomuser.GetUser()}"} - async with session.get(url, headers=headers) as response: - if response.status == 200: - printer.success( - f"{len(url_set) + 1} Valid URL(s) found : {Style.BRIGHT}{url}{Style.RESET_ALL}" - ) - url_set.add(url) + url = f"https://{domain}/{path}" + headers = {"User-Agent": str(randomuser.GetUser())} + try: + async with session.get(url, headers=headers) as response: + if response.status == 200: + url_set.add(url) + printer.success( + f"{len(url_set)} Valid URL(s) found : {Style.BRIGHT}{url}{Style.RESET_ALL}" + ) + except Exception: + pass -async def scan_async(paths: set[str]) -> None: +async def scan_async(domain: str, paths: set[str]) -> set[str]: """ - Scans the url asynchronously + Scans the domain asynchronously for all paths. + :param domain: target domain :param paths: set of paths to scan + :return: set of valid URLs found """ + url_set: set[str] = set() async with aiohttp.ClientSession() as session: - tasks = [fetch_url(session, path) for path in paths] + tasks = [fetch_url(session, domain, path, url_set) for path in paths] await asyncio.gather(*tasks, return_exceptions=True) + return url_set -def scan_urls() -> None: +def scan_urls(domain: str) -> set[str]: + """ + Loads the wordlist and runs the async scan. + + :param domain: target domain + :return: set of valid URLs found + """ paths = get_wordlist() - printer.debug(target_domain, paths) + printer.debug(f"Domain: {domain}, paths loaded: {len(paths) if paths else 0}") if paths is None: - printer.error("Connection Error..!") - return + printer.error("Failed to load wordlist..!") + return set() try: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - loop.run_until_complete(scan_async(paths)) + return asyncio.run(scan_async(domain, paths)) except KeyboardInterrupt: printer.error("Cancelled..!") + return set() except RuntimeError as e: - printer.error("Ran into a RuntimeError:", e) + printer.error(f"Ran into a RuntimeError: {e}") + return set() diff --git a/utils/email_search.py b/utils/email_search.py index 552f526..fc50067 100644 --- a/utils/email_search.py +++ b/utils/email_search.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by diff --git a/utils/fake_info_generator.py b/utils/fake_info_generator.py index dadf6f4..2323acf 100644 --- a/utils/fake_info_generator.py +++ b/utils/fake_info_generator.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by diff --git a/utils/ig_scrape.py b/utils/ig_scrape.py index 9b1e6ec..87d6475 100644 --- a/utils/ig_scrape.py +++ b/utils/ig_scrape.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -43,6 +43,10 @@ def scrape(target: str) -> None: printer.error("You are being rate limited..!") except APIError as e: printer.error("There is a issue with the API:", e) + except AttributeError: + printer.error( + "Couldn't get any data, you may need to change IP's or disable your VPN for a while..!" + ) def print_scraped_data(data: dict) -> None: diff --git a/utils/ip_lookup.py b/utils/ip_lookup.py index cd7db25..3986a1f 100644 --- a/utils/ip_lookup.py +++ b/utils/ip_lookup.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -15,7 +15,6 @@ along with this program. If not, see . """ -import json import socket import time @@ -30,35 +29,37 @@ def lookup(ip_address: str) -> None: """ Gets information about a given ip address using https://ipinfo.io/ - :param ip_address: The IP address to search for. + :param ip_address: The IP address or hostname to look up. """ try: ip_address = socket.gethostbyname(ip_address) url = f"https://ipinfo.io/{ip_address}/json" - headers = {"User-Agent": f"{randomuser.GetUser()}"} - url = requests.get(url, headers=headers) - # printer.info(url.text) - values = json.loads(url.text) + headers = {"User-Agent": str(randomuser.GetUser())} + response = requests.get(url, headers=headers) + response.raise_for_status() + values = response.json() printer.info( f"Trying to find information for {Style.BRIGHT}{ip_address}{Style.RESET_ALL}..." ) time.sleep(1) - for value in values: - # If value contains readme, skip it. - if value == "readme": + for key, value in values.items(): + if key == "readme": continue - elif value == "" or value is None: + if not value: value = "Not Found" + printer.success(f"{key.capitalize()} :", value) - printer.success(f"{value.capitalize()} :", values[value]) - - printer.success( - "Openstreetmap URL :", - f"https://www.openstreetmap.org/search?query={values['loc']}", - ) + if "loc" in values: + printer.success( + "Openstreetmap URL :", + f"https://www.openstreetmap.org/search?query={values['loc']}", + ) + except requests.exceptions.RequestException as e: + printer.error(f"Request error : {e}") + except socket.gaierror as e: + printer.error(f"Could not resolve host : {e}") except Exception as e: printer.error(f"Error : {e}") - pass diff --git a/utils/leak_search.py b/utils/leak_search.py index f9936d5..d034c05 100644 --- a/utils/leak_search.py +++ b/utils/leak_search.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -59,4 +59,3 @@ def lookup(target: str) -> None: except requests.exceptions.RequestException as e: printer.error(f"Error or the target wasn't found : {e}") - pass diff --git a/utils/local_users.py b/utils/local_users.py index 8016b8c..48d4d29 100644 --- a/utils/local_users.py +++ b/utils/local_users.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -95,6 +95,8 @@ def scan_for_local_users() -> None: try: user_info_list = [] + login_name = getpass.getuser() + for user in pwd.getpwall(): username = user.pw_name uid = user.pw_uid @@ -103,9 +105,8 @@ def scan_for_local_users() -> None: home_dir = user.pw_dir shell = user.pw_shell - # Get additional information using grp and getpass + # Get additional information using grp group_name = grp.getgrgid(gid)[0] - login_name = getpass.getuser() user_info = { "Username": username, diff --git a/utils/phonenumber_lookup.py b/utils/phonenumber_lookup.py index 7d58dc3..6a99d13 100644 --- a/utils/phonenumber_lookup.py +++ b/utils/phonenumber_lookup.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -33,7 +33,7 @@ def lookup(phone_number: str) -> None: """ try: ph_no = p.parse(phone_number) - country = p.region_code_for_country_code(ph_no.country_code if None else 0) + country = p.region_code_for_country_code(ph_no.country_code or 0) no_carrier = carrier.name_for_number(ph_no, "en") no_valid = p.is_valid_number(ph_no) no_possible = p.is_possible_number(ph_no) @@ -52,4 +52,4 @@ def lookup(phone_number: str) -> None: printer.success("Region -", region) printer.success("Time Zone -", time_zone) except Exception as e: - printer.error("Error : ", e) + printer.error(f"Error : {e}") diff --git a/utils/port_scanner.py b/utils/port_scanner.py index 8cecb6a..dce0865 100644 --- a/utils/port_scanner.py +++ b/utils/port_scanner.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -22,9 +22,6 @@ from helper import printer, timer -open_ports = [] -failed_ports = [] - @timer.timer(require_input=True) def scan(ip: str, port_range: int) -> None: @@ -34,68 +31,88 @@ def scan(ip: str, port_range: int) -> None: :param ip: IP address. :param port_range: The range of ports to scan. """ + # Initialise fresh state for each run so consecutive scans don't accumulate. + open_ports: list[int] = [] + failed_ports: list[int] = [] + try: printer.info( - f"Scanning for open ports for {Style.BRIGHT}{ip}{Style.RESET_ALL} with the port range of {Style.BRIGHT}1-{port_range}{Style.RESET_ALL}..." + f"Scanning for open ports for {Style.BRIGHT}{ip}{Style.RESET_ALL} " + f"with the port range of {Style.BRIGHT}1-{port_range}{Style.RESET_ALL}..." ) if port_range > 1000: printer.warning("This may take a while...") - scan_ports(ip, port_range) + scan_ports(ip, port_range, open_ports, failed_ports) - if len(open_ports) == 0: + if not open_ports: printer.error( f"No open ports found for {Style.BRIGHT}{ip}{Style.RESET_ALL}..!" ) else: printer.success( - f"Found {len(open_ports)}/{len(failed_ports)} open ports in '{ip}'..!" + f"Found {len(open_ports)}/{len(open_ports) + len(failed_ports)} " + f"open ports in '{ip}'..!" ) except KeyboardInterrupt: printer.error("Cancelled..!") except RecursionError: - printer.error("wtf.") + printer.error("Unexpected recursion error.") -def scan_ports(ip: str, port_range: int) -> None: +def scan_ports( + ip: str, + port_range: int, + open_ports: list[int], + failed_ports: list[int], +) -> None: """ - Scans for open ports in a given IP address. + Scans for open ports in a given IP address using a thread pool. :param ip: IP address. :param port_range: The range of ports to scan. + :param open_ports: List to collect open port numbers into. + :param failed_ports: List to collect closed/timed-out port numbers into. """ with ThreadPoolExecutor(max_workers=50) as executor: futures = { - executor.submit(connect_to_port, ip, port): port + executor.submit(connect_to_port, ip, port, open_ports, failed_ports): port for port in range(1, port_range + 1) } for future in as_completed(futures): - result = future.result() - if result is not None: - printer.success(result) + # Exceptions are already handled inside connect_to_port; just drain results. + future.result() -def connect_to_port(ip: str, port: int) -> None: +def connect_to_port( + ip: str, + port: int, + open_ports: list[int], + failed_ports: list[int], +) -> None: """ - Scans an individual port of a given IP address. + Attempts to connect to a single port on the given IP address. + Records the result in open_ports or failed_ports accordingly. :param ip: IP address. :param port: Port number. - :return: Success message if port is open, None otherwise. + :param open_ports: Shared list to append open ports to. + :param failed_ports: Shared list to append failed ports to. """ try: with socket.socket() as sock: sock.settimeout(0.5) - sock.connect((str(ip), port)) + sock.connect((ip, port)) open_ports.append(port) - return printer.success( - f"Found a open port : {Style.BRIGHT}{port}{Style.RESET_ALL}" + printer.success( + f"Found an open port : {Style.BRIGHT}{port}{Style.RESET_ALL}" ) except socket.timeout: failed_ports.append(port) except ConnectionRefusedError: - return None + failed_ports.append(port) except socket.error as e: - return printer.error( - f"An error occurred while scanning port {Style.BRIGHT}{port}{Style.RESET_ALL} for {Style.BRIGHT}{ip}{Style.RESET_ALL} : {str(e)}" + printer.error( + f"An error occurred while scanning port {Style.BRIGHT}{port}{Style.RESET_ALL} " + f"for {Style.BRIGHT}{ip}{Style.RESET_ALL} : {e}" ) diff --git a/utils/search_username.py b/utils/search_username.py index 02c5acb..0a9ce04 100644 --- a/utils/search_username.py +++ b/utils/search_username.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -18,7 +18,6 @@ import asyncio import json from datetime import datetime -from typing import Any import aiohttp from colorama import Style @@ -37,31 +36,34 @@ def search(username: str) -> None: check_user_from_data(username) except KeyboardInterrupt: printer.error("Cancelled..!") - pass -def check_user_from_data(username: str) -> dict[str, dict[str, str | int] | list[Any]]: +def check_user_from_data(username: str) -> dict: """ Scans for the given username across many different sites. :param username: The username to scan for. + :return: A dict summarising the search parameters and matched sites. """ + # Read the data file once and reuse it throughout this call. + data = url_helper.read_local_content("resources/data.json") + if not isinstance(data, dict): + printer.error("Failed to load site data.") + return {} + + sites = data.get("sites", []) printer.info( - f"Searching for {Style.BRIGHT}{username}{Style.RESET_ALL} across {len(url_helper.read_local_content('resources/data.json')['sites'])} different websites..." + f"Searching for {Style.BRIGHT}{username}{Style.RESET_ALL} " + f"across {len(sites)} different websites..." ) - results = [] - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - loop.run_until_complete(make_requests(username)) + results = asyncio.run(make_requests(username, sites)) now = datetime.now().strftime("%m/%d/%Y %H:%M:%S") user_json = { "search-params": { "username": username, - "sites-number": len( - url_helper.read_local_content("resources/data.json")["sites"] - ), + "sites-number": len(sites), "date": now, }, "sites": results, @@ -70,33 +72,50 @@ def check_user_from_data(username: str) -> dict[str, dict[str, str | int] | list return user_json -async def make_requests(username: str) -> None: +async def make_requests(username: str, sites: list) -> list: """ - Makes the requests to the sites. + Makes the requests to all sites and returns a list of matched results. :param username: The username to scan for. + :param sites: List of site configuration dicts from data.json. + :return: List of site dicts where the username was found. """ async with aiohttp.ClientSession( timeout=aiohttp.ClientTimeout(total=20) ) as session: - tasks = [] - for content in url_helper.read_local_content("resources/data.json")["sites"]: - task = asyncio.ensure_future(make_request(session, content, username)) - tasks.append(task) - await asyncio.gather(*tasks) + tasks = [ + asyncio.ensure_future(make_request(session, content, username)) + for content in sites + ] + results = await asyncio.gather(*tasks) + + # Filter out None entries (sites where the username was not found). + return [r for r in results if r is not None] async def make_request( session: aiohttp.ClientSession, content: dict, username: str -) -> None: +) -> dict | None: + """ + Makes a single request to one site and returns the site dict on a match. + + :param session: The shared aiohttp client session. + :param content: Site configuration dict from data.json. + :param username: The username to check. + :return: The site dict if the username was found, otherwise None. + """ url = content["url"].format(username=username) json_body = None - headers = {"User-Agent": f"{randomuser.GetUser()}"} + headers = {"User-Agent": str(randomuser.GetUser())} + if "headers" in content: - headers.update(eval(content["headers"])) + # NOTE: eval() is used here because the data.json format stores headers + # as Python expression strings. Treat data.json as trusted input only. + headers.update(eval(content["headers"])) # noqa: S307 + if "json" in content: - json_body = content["json"].format(username=username) - json_body = json.loads(json_body) + json_body = json.loads(content["json"].format(username=username)) + try: async with session.request( content["method"], @@ -106,9 +125,15 @@ async def make_request( headers=headers, ssl=False, ) as response: - if eval(content["valid"]): + # `valid` is a Python expression string evaluated against the response. + # NOTE: same caveat as above — data.json must be trusted. + if eval(content["valid"]): # noqa: S307 printer.success( - f"#{content['id']} {Style.BRIGHT}{content['app']}{Style.RESET_ALL} - {url} [{response.status} {response.reason}]" + f"#{content['id']} {Style.BRIGHT}{content['app']}{Style.RESET_ALL}" + f" - {url} [{response.status} {response.reason}]" ) + return content except Exception: pass + + return None diff --git a/utils/web_scrape.py b/utils/web_scrape.py index 94bdab1..9f573b4 100644 --- a/utils/web_scrape.py +++ b/utils/web_scrape.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -20,7 +20,6 @@ import json from datetime import datetime from pathlib import Path -from typing import Any from urllib.parse import urljoin, urlparse import aiohttp @@ -29,8 +28,6 @@ from helper import printer, randomuser, timer -scraped_links = set() - def export_links(links: set, base_url: str, format_type: str = "txt") -> None: """ @@ -117,8 +114,10 @@ def scrape(url: str) -> None: :param url: url of the website. """ - base_url = urlparse(url).netloc - printer.debug(f"Scraping {base_url}") + printer.debug(f"Scraping {urlparse(url).netloc}") + + # Use a fresh set for every invocation so state never leaks between runs. + scraped_links: set[str] = set() try: response = printer.user_input( @@ -131,14 +130,13 @@ def scrape(url: str) -> None: printer.warning( "This may take a while depending on the sizes of the sites." ) - - asyncio.run(scrape_links(url, recursive=True)) + asyncio.run(scrape_links(url, scraped_links, recursive=True)) printer.success("Scraping linked pages completed..!") else: printer.info( f"Trying to scrape links from {Style.BRIGHT}{url}{Style.RESET_ALL}..." ) - asyncio.run(scrape_links(url, recursive=False)) + asyncio.run(scrape_links(url, scraped_links, recursive=False)) printer.success("Scraping completed..!") # Ask user if they want to export the results @@ -166,30 +164,56 @@ def scrape(url: str) -> None: export_format = format_map.get(format_choice, "txt") export_links(scraped_links, url, export_format) - # Clear scraped links for next run - scraped_links.clear() - - except Exception as e: - printer.error(f"Error : {e}") except KeyboardInterrupt: printer.error("Cancelled..!") + except Exception as e: + printer.error(f"Error : {e}") + +async def fetch(session: aiohttp.ClientSession, url: str) -> str: + """ + Fetches the HTML content of a URL. -async def fetch(session, url: str) -> str: - headers = {"User-Agent": f"{randomuser.GetUser()}"} - async with session.get(url, headers=headers) as response: - return await response.text() + :param session: The shared aiohttp client session. + :param url: URL to fetch. + :return: Response body as text, or an empty string on error. + """ + headers = {"User-Agent": str(randomuser.GetUser())} + try: + async with session.get(url, headers=headers) as response: + return await response.text() + except Exception: + return "" -async def parse_links( - content: str, base_url: str -) -> list[tuple[str | bytes | Any, str]]: +async def parse_links(content: str, base_url: str) -> list[tuple[str, str]]: + """ + Parses all anchor tags from *content* and returns absolute (href, text) pairs. + + :param content: Raw HTML string. + :param base_url: Base URL used to resolve relative hrefs. + :return: List of (absolute_url, link_text) tuples. + """ soup = BeautifulSoup(content, "html.parser") - links = soup.find_all("a") - return [(urljoin(base_url, str(link.get("href"))), link.text) for link in links] + return [ + (urljoin(base_url, str(link.get("href"))), link.get_text(strip=True)) + for link in soup.find_all("a") + ] + +async def scrape_links( + url: str, + scraped_links: set[str], + recursive: bool = False, +) -> None: + """ + Scrapes links from *url* and, optionally, from every discovered page. -async def scrape_links(url: str | Any, recursive=False) -> None: + :param url: The URL to scrape. + :param scraped_links: Shared set used to track already-visited URLs across + recursive calls, preventing infinite loops. + :param recursive: When True, every newly discovered URL is scraped as well. + """ async with aiohttp.ClientSession() as session: html_content = await fetch(session, url) links = await parse_links(html_content, url) @@ -198,9 +222,9 @@ async def scrape_links(url: str | Any, recursive=False) -> None: if href not in scraped_links: scraped_links.add(href) printer.success( - f"{len(scraped_links)} Link(s) found : {Style.BRIGHT}{href} - {text}{Style.RESET_ALL}" + f"{len(scraped_links)} Link(s) found : " + f"{Style.BRIGHT}{href} - {text}{Style.RESET_ALL}" ) if recursive: - # await asyncio.sleep(0.5) - await scrape_links(href) # recursively scrape linked pages + await scrape_links(href, scraped_links, recursive=True) diff --git a/utils/websearch.py b/utils/websearch.py index c7de761..67e474b 100644 --- a/utils/websearch.py +++ b/utils/websearch.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -22,14 +22,18 @@ from helper import printer, randomuser, timer -headers = { - "User-Agent": f"{randomuser.GetUser()}", +_STATIC_HEADERS = { "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.5", "Referer": "https://duckduckgo.com/", } +def _build_headers() -> dict: + """Return a fresh headers dict with a newly sampled User-Agent.""" + return {**_STATIC_HEADERS, "User-Agent": str(randomuser.GetUser())} + + @timer.timer(require_input=True) def websearch(query: str) -> None: """ @@ -57,7 +61,7 @@ def send_request(url: str) -> Response | None: :return: The response object if successful, or None. """ try: - with requests.get(url, headers=headers) as response: + with requests.get(url, headers=_build_headers()) as response: response.raise_for_status() return response except requests.exceptions.RequestException: @@ -82,10 +86,10 @@ def parse_and_print_results(response_text: str, query: str) -> None: if any(keyword in query for keyword in dork_keywords): printer.info(f"Searching with dorks {Style.BRIGHT}{query}{Style.RESET_ALL}") - printer.debug(headers["User-Agent"]) else: printer.info(f"Searching for {Style.BRIGHT}{query}{Style.RESET_ALL}") - printer.debug(headers["User-Agent"]) + + printer.debug(f"User-Agent: {_build_headers()['User-Agent']}") for result in results: print_search_result(result) @@ -105,7 +109,7 @@ def print_search_result(result) -> None: f"{Style.BRIGHT}{title}{Style.RESET_ALL} : {link} \t[{status_code}]" ) except Exception as e: - printer.error(e) + printer.error(f"Error parsing result: {e}") def get_status_code(url: str) -> int | None: @@ -116,7 +120,9 @@ def get_status_code(url: str) -> int | None: :return: The status code if the request is successful, or None otherwise. """ try: - with requests.head(url, allow_redirects=True) as response: + with requests.head( + url, headers=_build_headers(), allow_redirects=True + ) as response: response.raise_for_status() return response.status_code except requests.exceptions.RequestException: diff --git a/utils/whois_lookup.py b/utils/whois_lookup.py index 0b84be3..6fdff4d 100644 --- a/utils/whois_lookup.py +++ b/utils/whois_lookup.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by diff --git a/utils/wifi_finder.py b/utils/wifi_finder.py index a6f5b1a..14a2bda 100644 --- a/utils/wifi_finder.py +++ b/utils/wifi_finder.py @@ -1,5 +1,5 @@ """ -Copyright (c) 2023-2025. Vili and contributors. +Copyright (c) 2023-2026. Vili and contributors. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -45,7 +45,7 @@ def scan_windows() -> None: ) try: output = subprocess.check_output(["netsh", "wlan", "show", "networks"]) - parse_output(output.decode("utf-8"), "windows") + parse_windows_output(output.decode("utf-8")) except subprocess.CalledProcessError as e: printer.error(f"Error : {e.returncode} - {e.stderr}") @@ -55,45 +55,125 @@ def scan_linux() -> None: f"Linux system detected... Performing {Style.BRIGHT}nmcli{Style.RESET_ALL} scan..." ) try: - output = subprocess.check_output(["nmcli", "dev", "wifi"]) - parse_output(output.decode("utf-8"), "linux") + # Use terse mode (-t) with explicit fields so the output is easy to split + # reliably regardless of column widths or terminal size. + # Fields: IN-USE, SSID, SIGNAL, SECURITY + output = subprocess.check_output( + ["nmcli", "-t", "-f", "IN-USE,SSID,SIGNAL,SECURITY", "dev", "wifi"] + ) + parse_linux_output(output.decode("utf-8")) + except FileNotFoundError: + printer.error( + f"Could not find {Style.BRIGHT}nmcli{Style.RESET_ALL}. " + "Is NetworkManager installed on your system?" + ) except subprocess.CalledProcessError as e: printer.error(f"Error : {e.returncode} - {e.stderr}") printer.error(f"Is your system using {Style.BRIGHT}nmcli{Style.RESET_ALL}?") -def parse_output(output: str, platform: str) -> None: - match platform: - case "windows": - # Parse Windows output - networks = [] - for line in output.splitlines(): - if "SSID" in line: - parts = line.split(":") - if len(parts) > 1: - ssid = parts[1].strip() - networks.append({"ssid": ssid, "signal": "", "encryption": ""}) - printer.info("Available Wi-Fi networks :") - for network in networks: - printer.success( - f" {network['ssid']} (Signal: {network['signal']}, Encryption: {network['encryption']})" - ) - case "linux": - # Parse Linux output - networks = [] - for line in output.splitlines(): - if "*" in line: - parts = line.split() - ssid = " ".join(parts[1:-3]) # Extract Wi-Fi name - signal = parts[-3] - encryption = parts[-2] - networks.append( - {"ssid": ssid, "signal": signal, "encryption": encryption} - ) - printer.info("Available Wi-Fi networks :") - for network in networks: - printer.success( - f" {network['ssid']} (Signal: {network['signal']}, Encryption: {network['encryption']})" - ) - case _: - printer.error("idk how u got here.") +def parse_windows_output(output: str) -> None: + """ + Parses the output of `netsh wlan show networks` and prints each network. + + :param output: Raw decoded output from netsh. + """ + networks: list[dict] = [] + current: dict = {} + + for line in output.splitlines(): + line = line.strip() + if line.startswith("SSID") and "BSSID" not in line: + # Start of a new network block. + if current: + networks.append(current) + parts = line.split(":", 1) + current = { + "ssid": parts[1].strip() if len(parts) > 1 else "", + "signal": "", + "encryption": "", + } + elif line.startswith("Signal"): + parts = line.split(":", 1) + current["signal"] = parts[1].strip() if len(parts) > 1 else "" + elif line.startswith("Authentication"): + parts = line.split(":", 1) + current["encryption"] = parts[1].strip() if len(parts) > 1 else "" + + if current: + networks.append(current) + + if not networks: + printer.warning("No Wi-Fi networks found.") + return + + printer.info(f"Available Wi-Fi networks ({len(networks)} found) :") + for network in networks: + printer.success( + f" {network['ssid']}" + f" (Signal: {network['signal'] or 'N/A'}," + f" Encryption: {network['encryption'] or 'N/A'})" + ) + + +def parse_linux_output(output: str) -> None: + """ + Parses the terse output of `nmcli -t -f IN-USE,SSID,SIGNAL,SECURITY dev wifi` + and prints every visible network, marking the connected one with an asterisk. + + Terse format fields are separated by ':'. Literal colons inside field values + are escaped as '\\:' by nmcli. + + :param output: Raw decoded terse nmcli output. + """ + networks: list[dict] = [] + + for line in output.splitlines(): + line = line.strip() + if not line: + continue + + # Split on unescaped colons only. + # nmcli escapes literal colons inside values as '\:', so we split on ':' + # that is NOT preceded by a backslash. + import re + + parts = re.split(r"(?