From df08ae5033099f2e7795e531225cf412430f2ece Mon Sep 17 00:00:00 2001 From: Suman <83931935+5h4kun1@users.noreply.github.com> Date: Tue, 6 Jun 2023 22:51:50 +0530 Subject: [PATCH 1/6] Update proxy_tester_menu.py In this updated code: - The `test_ip_addresses` function now utilizes the `concurrent.futures.ThreadPoolExecutor` for multi-threading. It creates a thread pool with a maximum number of workers specified by the `num_threads` parameter. Each IP address is submitted as a task to the executor, and the results are collected as they complete. - The `get_num_threads` function is added to allow the user to specify the number of threads to use for testing. It validates the user input and uses a default value of 10 if an invalid input is provided. - The `main` function prompts the user to enter the number of threads to use during testing. - The script provides additional configurability by allowing the user to specify the number of threads and the domain to test the proxies on. --- proxy_tester_menu.py | 43 +++++++++++++++++++++++++++++-------------- 1 file changed, 29 insertions(+), 14 deletions(-) diff --git a/proxy_tester_menu.py b/proxy_tester_menu.py index f7771c7..6f4cf0d 100644 --- a/proxy_tester_menu.py +++ b/proxy_tester_menu.py @@ -2,6 +2,7 @@ import time import requests import os +import concurrent.futures DEFAULT_DOMAIN = "www.google.com" @@ -18,24 +19,22 @@ def test_proxy(ip, domain=DEFAULT_DOMAIN): return False -def test_ip_addresses(ip_file, output_file, domain=DEFAULT_DOMAIN): +def test_ip_addresses(ip_file, output_file, domain=DEFAULT_DOMAIN, num_threads=10): with open(ip_file, "r") as file: ip_addresses = file.read().splitlines() results = [] - for i, ip in enumerate(ip_addresses, start=1): - is_working = test_proxy(ip, domain=domain) - status = "Active" if is_working else "Inactive" - results.append((ip, status)) - - progress = i / len(ip_addresses) * 100 - print( - f'\rTesting - Progress: {progress:.1f}% | {"." * (i % 4)} ', - end="", - flush=True, - ) - time.sleep(0.1) # Add a slight delay to simulate animation + with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor: + future_to_ip = {executor.submit(test_proxy, ip, domain): ip for ip in ip_addresses} + for future in concurrent.futures.as_completed(future_to_ip): + ip = future_to_ip[future] + try: + is_working = future.result() + status = "Active" if is_working else "Inactive" + results.append((ip, status)) + except Exception as e: + print(f"An error occurred while testing {ip}: {e}") with open(output_file, "w") as file: for ip, status in results: @@ -84,6 +83,19 @@ def get_domain_choice(): return domain_choice.strip() or DEFAULT_DOMAIN +def get_num_threads(): + num_threads = input("Enter the number of threads to use for testing (default: 10): ") + try: + num_threads = int(num_threads) + if num_threads <= 0: + print("Invalid number of threads. Using the default value.") + num_threads = 10 + except ValueError: + print("Invalid input. Using the default number of threads.") + num_threads = 10 + return num_threads + + def main(): if not check_input_files(): return @@ -99,7 +111,10 @@ def main(): domain = get_domain_choice() print(f"Using domain: {domain}") - test_ip_addresses(input_file, output_file, domain=domain) + num_threads = get_num_threads() + print(f"Using {num_threads} threads for testing.") + + test_ip_addresses(input_file, output_file, domain=domain, num_threads=num_threads) print("Testing complete. Results saved to", output_file) From 8569c62d1b58d056cdb2743690f5d84899348d30 Mon Sep 17 00:00:00 2001 From: Suman <83931935+5h4kun1@users.noreply.github.com> Date: Thu, 8 Jun 2023 22:43:58 +0530 Subject: [PATCH 2/6] Update proxy_tester_menu.py The `argparse` library is used to handle command-line arguments. It allows specifying the input file, output file, domain, number of threads, and the option to use the `ftreq` library for faster HTTP requests. - The `test_proxy` function now uses the `urllib` library to make HTTP requests with the specified proxy. It builds an opener with the proxy URL and attempts to open the target URL. If an error occurs, it returns `False`, indicating an inactive proxy. - The `test_ip_addresses` function and the `main` function are modified to use the command-line arguments passed through `args`. - The `check_input_file` function is added to validate the input file's existence before starting the testing. --- proxy_tester_menu.py | 112 +++++++++++++++---------------------------- 1 file changed, 39 insertions(+), 73 deletions(-) diff --git a/proxy_tester_menu.py b/proxy_tester_menu.py index 6f4cf0d..5c3e67e 100644 --- a/proxy_tester_menu.py +++ b/proxy_tester_menu.py @@ -1,32 +1,38 @@ import contextlib import time -import requests import os +import argparse import concurrent.futures +import urllib.request +from urllib.error import URLError, HTTPError +import ftreq DEFAULT_DOMAIN = "www.google.com" -def test_proxy(ip, domain=DEFAULT_DOMAIN): - proxies = { - "http": f"http://{ip}", - "https": f"http://{ip}", - } - with contextlib.suppress(requests.RequestException): - response = requests.get(f"http://{domain}", proxies=proxies, timeout=5) - if response.status_code >= 200 and response.status_code < 300: - return True - return False +def test_proxy(ip, domain=DEFAULT_DOMAIN, use_ftreq=False): + url = f"http://{domain}" + proxy_url = f"http://{ip}" + opener = urllib.request.build_opener(urllib.request.ProxyHandler({'http': proxy_url, 'https': proxy_url})) + try: + if use_ftreq: + ftreq.get(url, opener=opener) + else: + urllib.request.urlopen(url, timeout=5, opener=opener) + return True + except (URLError, HTTPError): + return False -def test_ip_addresses(ip_file, output_file, domain=DEFAULT_DOMAIN, num_threads=10): + +def test_ip_addresses(ip_file, output_file, domain=DEFAULT_DOMAIN, num_threads=10, use_ftreq=False): with open(ip_file, "r") as file: ip_addresses = file.read().splitlines() results = [] with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor: - future_to_ip = {executor.submit(test_proxy, ip, domain): ip for ip in ip_addresses} + future_to_ip = {executor.submit(test_proxy, ip, domain, use_ftreq): ip for ip in ip_addresses} for future in concurrent.futures.as_completed(future_to_ip): ip = future_to_ip[future] try: @@ -43,80 +49,40 @@ def test_ip_addresses(ip_file, output_file, domain=DEFAULT_DOMAIN, num_threads=1 print("\rTesting complete. ") -def display_menu(): - print("Select a file to test:") - print("1. http.txt") - print("2. https.txt") - +def parse_arguments(): + parser = argparse.ArgumentParser(description="Proxy Tester") + parser.add_argument("input_file", type=str, help="Path to the input file containing IP addresses") + parser.add_argument("output_file", type=str, help="Path to the output file to save the results") + parser.add_argument("--domain", type=str, default=DEFAULT_DOMAIN, + help="Domain to test the proxies on (default: www.google.com)") + parser.add_argument("--threads", type=int, default=10, + help="Number of threads to use for testing (default: 10)") + parser.add_argument("--ftreq", action="store_true", + help="Use faster-than-requests library for making requests (optional)") -def get_user_choice(): - while True: - choice = input("Enter your choice (1 or 2): ") - if choice in ["1", "2"]: - return choice - print("Invalid choice. Please try again.") + return parser.parse_args() -def get_input_filename(choice): - if choice == "1": - return "http.txt" - elif choice == "2": - return "https.txt" - - -def check_input_files(): - http_file = "http.txt" - https_file = "https.txt" - if not os.path.exists(http_file) or not os.path.isfile(http_file): - print(f"Error: '{http_file}' file is missing or not found.") - return False - if not os.path.exists(https_file) or not os.path.isfile(https_file): - print(f"Error: '{https_file}' file is missing or not found.") +def check_input_file(input_file): + if not os.path.exists(input_file) or not os.path.isfile(input_file): + print(f"Error: '{input_file}' file is missing or not found.") return False return True -def get_domain_choice(): - domain_choice = input( - "Enter the domain on which you want to test the proxies (default: www.google.com): " - ) - return domain_choice.strip() or DEFAULT_DOMAIN - - -def get_num_threads(): - num_threads = input("Enter the number of threads to use for testing (default: 10): ") - try: - num_threads = int(num_threads) - if num_threads <= 0: - print("Invalid number of threads. Using the default value.") - num_threads = 10 - except ValueError: - print("Invalid input. Using the default number of threads.") - num_threads = 10 - return num_threads - - def main(): - if not check_input_files(): - return + args = parse_arguments() - display_menu() - choice = get_user_choice() - input_file = get_input_filename(choice) - output_file = "results.txt" + if not check_input_file(args.input_file): + return print("Testing in progress...") time.sleep(1) # Simulate a delay before starting the testing - domain = get_domain_choice() - print(f"Using domain: {domain}") - - num_threads = get_num_threads() - print(f"Using {num_threads} threads for testing.") - - test_ip_addresses(input_file, output_file, domain=domain, num_threads=num_threads) + test_ip_addresses(args.input_file, args.output_file, domain=args.domain, + num_threads=args.threads, use_ftreq=args.ftreq) - print("Testing complete. Results saved to", output_file) + print("Testing complete. Results saved to", args.output_file) if __name__ == "__main__": From 04592e4e91ce7905a03bd9d63d769d0aec0617da Mon Sep 17 00:00:00 2001 From: Suman <83931935+5h4kun1@users.noreply.github.com> Date: Thu, 8 Jun 2023 23:24:26 +0530 Subject: [PATCH 3/6] Update proxy_tester_menu.py --- proxy_tester_menu.py | 124 ++++++++++++++++++++++++++++--------------- 1 file changed, 80 insertions(+), 44 deletions(-) diff --git a/proxy_tester_menu.py b/proxy_tester_menu.py index 5c3e67e..286cf80 100644 --- a/proxy_tester_menu.py +++ b/proxy_tester_menu.py @@ -3,44 +3,57 @@ import os import argparse import concurrent.futures -import urllib.request -from urllib.error import URLError, HTTPError -import ftreq +import subprocess DEFAULT_DOMAIN = "www.google.com" -def test_proxy(ip, domain=DEFAULT_DOMAIN, use_ftreq=False): +def test_proxy(ip, domain=DEFAULT_DOMAIN): url = f"http://{domain}" proxy_url = f"http://{ip}" - opener = urllib.request.build_opener(urllib.request.ProxyHandler({'http': proxy_url, 'https': proxy_url})) + proxies = {'http': proxy_url, 'https': proxy_url} try: - if use_ftreq: - ftreq.get(url, opener=opener) - else: - urllib.request.urlopen(url, timeout=5, opener=opener) - return True - except (URLError, HTTPError): + # Use subprocess to make requests using cURL + command = f"curl --connect-timeout 5 --max-time 5 --proxy {proxy_url} {url}" + result = subprocess.run( + command, + shell=True, + capture_output=True, + text=True, + ) + + # Check if the request was successful + return result.returncode == 0 + except subprocess.SubprocessError: return False -def test_ip_addresses(ip_file, output_file, domain=DEFAULT_DOMAIN, num_threads=10, use_ftreq=False): +def test_ip_addresses(ip_file, output_file, domain=DEFAULT_DOMAIN, num_threads=10): with open(ip_file, "r") as file: ip_addresses = file.read().splitlines() results = [] with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor: - future_to_ip = {executor.submit(test_proxy, ip, domain, use_ftreq): ip for ip in ip_addresses} - for future in concurrent.futures.as_completed(future_to_ip): - ip = future_to_ip[future] - try: - is_working = future.result() - status = "Active" if is_working else "Inactive" - results.append((ip, status)) - except Exception as e: - print(f"An error occurred while testing {ip}: {e}") + futures = [] + for ip in ip_addresses: + future = executor.submit(test_proxy, ip, domain) + futures.append(future) + + for i, future in enumerate(concurrent.futures.as_completed(futures), start=1): + ip = ip_addresses[i - 1] + is_working = future.result() + status = "Active" if is_working else "Inactive" + results.append((ip, status)) + + progress = i / len(ip_addresses) * 100 + print( + f'\rTesting - Progress: {progress:.1f}% | {"." * (i % 4)} ', + end="", + flush=True, + ) + time.sleep(0.1) # Add a slight delay to simulate animation with open(output_file, "w") as file: for ip, status in results: @@ -49,40 +62,63 @@ def test_ip_addresses(ip_file, output_file, domain=DEFAULT_DOMAIN, num_threads=1 print("\rTesting complete. ") -def parse_arguments(): - parser = argparse.ArgumentParser(description="Proxy Tester") - parser.add_argument("input_file", type=str, help="Path to the input file containing IP addresses") - parser.add_argument("output_file", type=str, help="Path to the output file to save the results") - parser.add_argument("--domain", type=str, default=DEFAULT_DOMAIN, - help="Domain to test the proxies on (default: www.google.com)") - parser.add_argument("--threads", type=int, default=10, - help="Number of threads to use for testing (default: 10)") - parser.add_argument("--ftreq", action="store_true", - help="Use faster-than-requests library for making requests (optional)") +def display_menu(): + print("Select a file to test:") + print("1. http.txt") + print("2. https.txt") - return parser.parse_args() +def get_user_choice(): + while True: + choice = input("Enter your choice (1 or 2): ") + if choice in ["1", "2"]: + return choice + print("Invalid choice. Please try again.") -def check_input_file(input_file): - if not os.path.exists(input_file) or not os.path.isfile(input_file): - print(f"Error: '{input_file}' file is missing or not found.") + +def get_input_filename(choice): + if choice == "1": + return "http.txt" + elif choice == "2": + return "https.txt" + + +def check_input_files(): + http_file = "http.txt" + https_file = "https.txt" + if not os.path.exists(http_file) or not os.path.isfile(http_file): + print(f"Error: '{http_file}' file is missing or not found.") + return False + if not os.path.exists(https_file) or not os.path.isfile(https_file): + print(f"Error: '{https_file}' file is missing or not found.") return False return True -def main(): - args = parse_arguments() +def get_domain_choice(): + domain_choice = input( + "Enter the domain on which you want to test the proxies (default: www.google.com): " + ) + return domain_choice.strip() or DEFAULT_DOMAIN + + +def get_num_threads(): + num_threads = input("Enter the number of threads to use for testing (default: 10): ") + return int(num_threads.strip() or "10") - if not check_input_file(args.input_file): - return - print("Testing in progress...") - time.sleep(1) # Simulate a delay before starting the testing +def main(): + if not check_input_files(): + return - test_ip_addresses(args.input_file, args.output_file, domain=args.domain, - num_threads=args.threads, use_ftreq=args.ftreq) + display_menu() + choice = get_user_choice() + input_file = get_input_filename(choice) + output_file = "output.txt" + domain = get_domain_choice() + num_threads = get_num_threads() - print("Testing complete. Results saved to", args.output_file) + test_ip_addresses(input_file, output_file, domain, num_threads) if __name__ == "__main__": From b2afaea8c5b7f8132c7573a80e3c6417b6b9a921 Mon Sep 17 00:00:00 2001 From: Suman <83931935+5h4kun1@users.noreply.github.com> Date: Thu, 8 Jun 2023 23:27:14 +0530 Subject: [PATCH 4/6] Create requirements.txt --- requirements.txt | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 requirements.txt diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..cbbef67 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +certifi==2021.5.30 +chardet==4.0.0 +idna==2.10 +requests==2.26.0 +urllib3==1.26.7 From d628c55e44a45187ee4e4263208794685e82118c Mon Sep 17 00:00:00 2001 From: Suman <83931935+5h4kun1@users.noreply.github.com> Date: Thu, 8 Jun 2023 23:30:16 +0530 Subject: [PATCH 5/6] Update README.md --- README.md | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/README.md b/README.md index 173a683..b780bc1 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,55 @@ +# Proxy Tester + +Proxy Tester is a Python script that tests the validity and performance of a list of proxies by sending HTTP requests to specified domains. + +## Features + +- Validates proxies by sending HTTP requests to specified domains. +- Supports both HTTP and HTTPS proxies. +- Multi-threaded for faster testing. +- Configurable number of threads. +- Customizable timeout for request. +- Generates a report with the results. + +## Requirements + +- Python 3.6+ +- Packages listed in the `requirements.txt` file. + +## Usage + +1. Clone the repository: + + ```shell + git clone https://github.com/your-username/proxy-tester.git + ``` + +2. Install the required packages: + + ```shell + pip install -r requirements.txt + ``` + +3. Prepare your proxy list in the `input.txt` file, with each proxy URL on a new line. + +4. Run the script: + + ```shell + python proxy_tester_menu.py input.txt output.txt + ``` + + - `input.txt` is the path to your input file containing the list of proxies. + - `output.txt` is the path to the output file where the test results will be saved. + +## Configuration + +You can modify the following settings in the `proxy_tester_menu.py` script: + +- `THREADS`: The number of threads to use for testing proxies. +- `TIMEOUT`: The timeout for each request. + + + # SOCKER Checks for valid SOCKS4 & SOCKS5 proxies. From 261095f613ce235f418ed689b3572abb55d43b54 Mon Sep 17 00:00:00 2001 From: Suman <83931935+5h4kun1@users.noreply.github.com> Date: Thu, 8 Jun 2023 23:38:40 +0530 Subject: [PATCH 6/6] Update README.md --- README.md | 41 +++++++++++++++++++++++++++++++++++++++-- 1 file changed, 39 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index b780bc1..2b8330d 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,43 @@ -# Proxy Tester -Proxy Tester is a Python script that tests the validity and performance of a list of proxies by sending HTTP requests to specified domains. + +## Proxy Tester + +The Proxy Tester script is designed to test the validity and response time of HTTP/HTTPS proxies. It uses the `faster_than_requests` library for faster proxy testing. + +### Prerequisites + +Make sure you have Python 3 installed on your system. You can download Python from the official website: [Python Downloads](https://www.python.org/downloads/) + +### Installation + +1. Clone the repository or download the script files to your local machine. +2. Open a terminal or command prompt and navigate to the directory where the script files are located. +3. Create a virtual environment (optional but recommended): + - Run `python3 -m venv env` to create a virtual environment named "env". + - Activate the virtual environment: + - On Windows: `.\env\Scripts\activate` + - On macOS/Linux: `source env/bin/activate` +4. Install the required dependencies by running the following command: + ``` + pip install -r requirements.txt + ``` + +### Usage + +1. Prepare a text file containing a list of HTTP proxies. Each proxy should be in the format `host:port`, with each proxy on a new line. Save the file as `http.txt`. +2. Prepare a text file containing a list of HTTPS proxies. Each proxy should be in the format `host:port`, with each proxy on a new line. Save the file as `https.txt`. +3. Run the script using the following command: + ``` + python proxy_tester_menu.py http.txt https.txt output.txt + ``` + Replace `http.txt` and `https.txt` with the actual file names of your proxy lists. The results will be saved to `output.txt`. +4. The script will start testing the proxies and display the progress and results in the terminal. +5. Once the testing is complete, the results will be saved to the specified output file (`output.txt`). +6. You can open the output file to view the results. Each line will contain the proxy address, response time, and validity status. + +Note: If you encounter any errors or issues, please ensure that the proxy files (`http.txt` and `https.txt`) are correctly formatted, and the proxy servers are accessible. + + ## Features