vulnerability_tester.py

import requests
from tqdm import tqdm
import time
from termcolor import colored
from tabulate import tabulate
import os

class VulnerabilityTester:
    def __init__(self, url):
        self.url = url
        self.results = []

    def load_payloads(self, filename):
        """Load payloads from external file."""
        try:
            with open(filename, 'r') as file:
                return [line.strip() for line in file.readlines()]
        except FileNotFoundError:
            print(colored(f"Error: {filename} not found!", "red"))
            return []
        except Exception as e:
            print(colored(f"Error loading payloads: {str(e)}", "red"))
            return []

    def run_tests(self):
        """Runs all tests (XSS, RFI, LFI, Command Injection, Open Redirect, Host Header Injection, CSRF, Subdomain Takeover, Clickjacking, SSRF, File Upload) sequentially."""
        self.test_xss()
        self.test_rfi()
        self.test_lfi()
        self.test_command_injection()
        self.test_open_redirect()
        self.test_host_header_injection()
        self.test_csrf()
        self.test_subdomain_takeover()
        self.test_clickjacking()
        self.test_ssrf()
        self.upload_file()
        self.display_results()

    def test_xss(self):
        """Testing for XSS vulnerability"""
        print(colored("Testing XSS...", "yellow"))
        payloads = self.load_payloads("xss.txt")
        for payload in tqdm(payloads, desc="XSS Testing"):
            full_url = f"{self.url}?input={payload}"
            try:
                response = requests.get(full_url)
                if payload in response.text:
                    self.results.append(["XSS", full_url, "Vulnerable"])
                else:
                    self.results.append(["XSS", full_url, "Safe"])
            except requests.exceptions.RequestException as e:
                self.results.append(["XSS", full_url, f"Error: {str(e)}"])
            time.sleep(0.03)  # Simulate delay

    def test_rfi(self):
        """Testing for Remote File Inclusion (RFI)"""
        print(colored("Testing RFI...", "yellow"))
        payloads = self.load_payloads("rfi.txt")
        for payload in tqdm(payloads, desc="RFI Testing"):
            full_url = f"{self.url}?file={payload}"
            try:
                response = requests.get(full_url)
                if "malicious.php" in response.text:
                    self.results.append(["RFI", full_url, "Vulnerable"])
                else:
                    self.results.append(["RFI", full_url, "Safe"])
            except requests.exceptions.RequestException as e:
                self.results.append(["RFI", full_url, f"Error: {str(e)}"])
            time.sleep(0.03)  # Simulate delay

    def test_lfi(self):
        """Testing for Local File Inclusion (LFI)"""
        print(colored("Testing LFI...", "yellow"))
        payloads = self.load_payloads("lfi.txt")
        for payload in tqdm(payloads, desc="LFI Testing"):
            full_url = f"{self.url}?file={payload}"
            try:
                response = requests.get(full_url)
                if "root:x" in response.text:
                    self.results.append(["LFI", full_url, "Vulnerable"])
                else:
                    self.results.append(["LFI", full_url, "Safe"])
            except requests.exceptions.RequestException as e:
                self.results.append(["LFI", full_url, f"Error: {str(e)}"])
            time.sleep(0.03)  # Simulate delay

    def test_command_injection(self):
        """Testing for Command Injection"""
        print(colored("Testing Command Injection...", "yellow"))
        payloads = self.load_payloads("cmd_injection.txt")
        for payload in tqdm(payloads, desc="Command Injection Testing"):
            full_url = f"{self.url}?cmd={payload}"
            try:
                response = requests.get(full_url)
                if "command result" in response.text:  # You can replace this with a better check
                    self.results.append(["Command Injection", full_url, "Vulnerable"])
                else:
                    self.results.append(["Command Injection", full_url, "Safe"])
            except requests.exceptions.RequestException as e:
                self.results.append(["Command Injection", full_url, f"Error: {str(e)}"])
            time.sleep(0.03)  # Simulate delay

    def test_open_redirect(self):
        """Testing for Open Redirect"""
        print(colored("Testing Open Redirect...", "yellow"))
        payloads = self.load_payloads("open_redirect.txt")
        for payload in tqdm(payloads, desc="Open Redirect Testing"):
            full_url = f"{self.url}?redirect={payload}"
            try:
                response = requests.get(full_url, allow_redirects=False)
                if response.status_code == 302 and payload in response.headers.get("Location", ""):
                    self.results.append(["Open Redirect", full_url, "Vulnerable"])
                else:
                    self.results.append(["Open Redirect", full_url, "Safe"])
            except requests.exceptions.RequestException as e:
                self.results.append(["Open Redirect", full_url, f"Error: {str(e)}"])
            time.sleep(0.03)  # Simulate delay

    def test_host_header_injection(self):
        """Testing for Host Header Injection"""
        print(colored("Testing Host Header Injection...", "yellow"))
        payloads = self.load_payloads("host_header_injection.txt")
        for payload in tqdm(payloads, desc="Host Header Injection Testing"):
            headers = {"Host": payload}
            try:
                response = requests.get(self.url, headers=headers)
                if payload in response.text or "malicious host" in response.text:  # Customize this check
                    self.results.append(["Host Header Injection", self.url, "Vulnerable"])
                else:
                    self.results.append(["Host Header Injection", self.url, "Safe"])
            except requests.exceptions.RequestException as e:
                self.results.append(["Host Header Injection", self.url, f"Error: {str(e)}"])
            time.sleep(0.03)  # Simulate delay

    def test_csrf(self):
        """Testing for CSRF vulnerability"""
        print(colored("Testing CSRF...", "yellow"))
        # Example CSRF attack simulation
        try:
            payload = f"<img src='{self.url}/change-password?new-password=malicious_password' />"
            response = requests.get(self.url)
            if "malicious_password" in response.text:  # Check if the password changed
                self.results.append(["CSRF", self.url, "Vulnerable"])
            else:
                self.results.append(["CSRF", self.url, "Safe"])
        except requests.exceptions.RequestException as e:
            self.results.append(["CSRF", self.url, f"Error: {str(e)}"])

    def test_subdomain_takeover(self):
        """Testing for Subdomain Takeover vulnerability"""
        print(colored("Testing Subdomain Takeover...", "yellow"))
        subdomains = ["test", "dev", "www"]  # List of potential subdomains
        for sub in tqdm(subdomains, desc="Subdomain Takeover Testing"):
            full_url = f"http://{sub}.{self.url}"
            try:
                response = requests.get(full_url)
                if response.status_code == 404:
                    self.results.append(["Subdomain Takeover", full_url, "Vulnerable (available for takeover)"])
                else:
                    self.results.append(["Subdomain Takeover", full_url, "Safe"])
            except requests.exceptions.RequestException as e:
                self.results.append(["Subdomain Takeover", full_url, f"Error: {str(e)}"])

    def test_clickjacking(self):
        """Testing for Clickjacking vulnerability"""
        print(colored("Testing Clickjacking...", "yellow"))
        try:
            headers = {"X-Frame-Options": "DENY"}
            response = requests.get(self.url, headers=headers)
            if "DENY" in response.text:  # Simple check for X-Frame-Options
                self.results.append(["Clickjacking", self.url, "Safe"])
            else:
                self.results.append(["Clickjacking", self.url, "Vulnerable"])
        except requests.exceptions.RequestException as e:
            self.results.append(["Clickjacking", self.url, f"Error: {str(e)}"])

    def test_ssrf(self):
        """Testing for Server-Side Request Forgery (SSRF)"""
        print(colored("Testing SSRF...", "yellow"))
        payloads = ["http://localhost/admin", "http://127.0.0.1:8000"]  # Example SSRF payloads
        for payload in tqdm(payloads, desc="SSRF Testing"):
            try:
                response = requests.get(self.url, params={"url": payload})
                if "admin panel" in response.text:  # Change this condition as per the application's response
                    self.results.append(["SSRF", self.url, "Vulnerable"])
                else:
                    self.results.append(["SSRF", self.url, "Safe"])
            except requests.exceptions.RequestException as e:
                self.results.append(["SSRF", self.url, f"Error: {str(e)}"])
            time.sleep(0.03)  # Simulate delay

    def upload_file(self):
        """Attempt to upload a malicious file for reverse shell"""
        print(colored("Uploading file...", "yellow"))
        malicious_file_path = "malicious.php"  # Path to your malicious file
        try:
            with open(malicious_file_path, 'rb') as file:
                files = {'file': file}
                response = requests.post(f"{self.url}/upload", files=files)  # Adjust the URL as needed
                if "Upload successful" in response.text:  # Check for a success message
                    self.results.append(["File Upload", self.url, "Vulnerable (Malicious file uploaded)"])
                else:
                    self.results.append(["File Upload", self.url, "Safe"])
        except requests.exceptions.RequestException as e:
            self.results.append(["File Upload", self.url, f"Error: {str(e)}"])
        except FileNotFoundError:
            self.results.append(["File Upload", self.url, "Error: Malicious file not found"])

    def display_results(self):
        """Display results in a tabular format"""
        print(colored("\nTesting Results:", "cyan"))
        print(tabulate(self.results, headers=["Test", "URL", "Result"], tablefmt="fancy_grid"))

def main():
    url = input("Enter the target URL: ")
    tester = VulnerabilityTester(url)
    tester.run_tests()

if __name__ == "__main__":
    main()