main.py

import requests
import pandas as pd
from urllib.parse import urljoin, urlparse
import socket
import ssl

def check_url_vulnerabilities(url, output_file="vulnerability_report.csv"):
    """
    Check if the given URL has common vulnerabilities and log the results to a CSV file.
    :param url: URL to check
    :param output_file: CSV file to save the results
    """
    print(f"Scanning {url} for vulnerabilities...\n")

    results = []

    def send_request(target_url):
        try:
            return requests.get(target_url, timeout=10)
        except requests.RequestException as e:
            print(f"[Error] Could not access {target_url}: {e}")
            return None

    # Check for SQL Injection
    sql_payloads = ["'", "\"", "1=1", "' OR '1'='1", '" OR "1"="1']
    for payload in sql_payloads:
        test_url = f"{url}?test={payload}"
        response = send_request(test_url)
        if response and ("SQL" in response.text or "syntax" in response.text):
            results.append({
                "URL": test_url,
                "Vulnerability Type": "SQL Injection",
                "Payload": payload
            })

    # Check for XSS
    xss_payloads = [
        "<script>alert('XSS')</script>",
        "<img src=x onerror=alert(1)>",
        "'><script>alert('XSS')</script>",
    ]
    for payload in xss_payloads:
        test_url = f"{url}?test={payload}"
        response = send_request(test_url)
        if response and payload in response.text:
            results.append({
                "URL": test_url,
                "Vulnerability Type": "XSS",
                "Payload": payload
            })

    # Check for Directory Traversal
    traversal_payloads = ["../../", "../etc/passwd", "../windows/system32"]
    for payload in traversal_payloads:
        test_url = urljoin(url, payload)
        response = send_request(test_url)
        if response and response.status_code == 200 and "root:" in response.text:
            results.append({
                "URL": test_url,
                "Vulnerability Type": "Directory Traversal",
                "Payload": payload
            })

    # Check for Open Directories
    response = send_request(url)
    if response and response.status_code == 200 and "Index of" in response.text:
        results.append({
            "URL": url,
            "Vulnerability Type": "Open Directory",
            "Payload": "N/A"
        })

    # Subdomain Discovery
    subdomains = ["admin", "test", "dev", "staging", "api"]
    parsed_url = urlparse(url)
    domain = parsed_url.netloc
    scheme = parsed_url.scheme
    for sub in subdomains:
        sub_url = f"{scheme}://{sub}.{domain}"
        try:
            response = requests.get(sub_url, timeout=5)
            if response.status_code == 200:
                results.append({
                    "URL": sub_url,
                    "Vulnerability Type": "Exposed Subdomain",
                    "Details": "Subdomain is accessible."
                })
        except requests.RequestException:
            pass

    # Port Scanning
    ports = [21, 22, 80, 443, 8080]
    hostname = domain.split(":")[0]
    for port in ports:
        try:
            with socket.create_connection((hostname, port), timeout=2):
                results.append({
                    "URL": f"{url}:{port}",
                    "Vulnerability Type": "Open Port",
                    "Details": f"Port {port} is open."
                })
        except (socket.timeout, ConnectionRefusedError, socket.gaierror):
            pass

    # Check Common Sensitive Endpoints
    endpoints = ["admin", "login", "config", ".env", "backup", "phpinfo.php"]
    for endpoint in endpoints:
        test_url = urljoin(url, endpoint)
        response = send_request(test_url)
        if response and response.status_code == 200:
            results.append({
                "URL": test_url,
                "Vulnerability Type": "Exposed Endpoint",
                "Details": f"Accessible endpoint: {endpoint}"
            })

    # Analyze HTTP Security Headers
    response = send_request(url)
    if response:
        security_headers = ["X-Frame-Options", "Content-Security-Policy", "X-Content-Type-Options", "Strict-Transport-Security"]
        for header in security_headers:
            if header not in response.headers:
                results.append({
                    "URL": url,
                    "Vulnerability Type": "Missing Security Header",
                    "Details": f"{header} is missing."
                })

    # Check for Command Injection
    command_payloads = ["; ls", "&& whoami", "| cat /etc/passwd"]
    for payload in command_payloads:
        test_url = f"{url}?input={payload}"
        response = send_request(test_url)
        if response and ("root:" in response.text or "bin/" in response.text):
            results.append({
                "URL": test_url,
                "Vulnerability Type": "Command Injection",
                "Details": f"Payload: {payload}"
            })

    # Check for CSRF Token
    response = send_request(url)
    if response and "<form" in response.text and "csrf" not in response.text.lower():
        results.append({
            "URL": url,
            "Vulnerability Type": "CSRF",
            "Details": "Forms found without anti-CSRF tokens."
        })

    # Check SSL/TLS Configuration
    try:
        with socket.create_connection((hostname, 443), timeout=5) as sock:
            with ssl.create_default_context().wrap_socket(sock, server_hostname=hostname) as ssock:
                cert = ssock.getpeercert()
                # Basic check for certificate expiration
                import datetime
                cert_not_after = cert['notAfter']
                cert_not_after_date = datetime.datetime.strptime(cert_not_after, "%b %d %H:%M:%S %Y %Z")
                if cert_not_after_date < datetime.datetime.now():
                    results.append({
                        "URL": url,
                        "Vulnerability Type": "SSL/TLS Configuration",
                        "Details": "Certificate has expired."
                    })
    except Exception as e:
        print(f"[Error] SSL/TLS check failed: {e}")

    # Check for Open Redirects
    redirect_payloads = ["/redirect?url=http://evil.com", "/login?next=http://evil.com"]
    for payload in redirect_payloads:
        test_url = urljoin(url, payload)
        try:
            response = requests.get(test_url, allow_redirects=False, timeout=5)
            if "Location" in response.headers and "http://evil.com" in response.headers["Location"]:
                results.append({
                    "URL": test_url,
                    "Vulnerability Type": "Open Redirect",
                    "Details": f"Payload: {payload}"
                })
        except requests.RequestException:
            pass

    # Check for File Upload Vulnerabilities
    # Note: This is a basic check and may not work for all scenarios.
    file_upload_url = urljoin(url, "/upload")
    files = {'file': ('test.php', '<?php echo "Vulnerable"; ?>', 'application/php')}
    try:
        response = requests.post(file_upload_url, files=files, timeout=5)
        if response.status_code == 200 and "Vulnerable" in response.text:
            results.append({
                "URL": file_upload_url,
                "Vulnerability Type": "File Upload",
                "Details": "Uploaded PHP file executed."
            })
    except requests.RequestException:
        pass

    # Directory Enumeration
    directories = ["admin", "backup", "test", "api", "config", "uploads"]
    for directory in directories:
        dir_url = urljoin(url, directory)
        response = send_request(dir_url)
        if response and response.status_code == 200:
            results.append({
                "URL": dir_url,
                "Vulnerability Type": "Exposed Directory",
                "Details": f"Directory found: {directory}"
            })

    # Check for Weak Session Management
    response = send_request(url)
    if response and "Set-Cookie" in response.headers:
        cookie_flags = response.headers["Set-Cookie"]
        if "HttpOnly" not in cookie_flags or "Secure" not in cookie_flags:
            results.append({
                "URL": url,
                "Vulnerability Type": "Weak Session Management",
                "Details": "Missing HttpOnly or Secure flags in cookies."
            })

    # Check for Exposed API Endpoints
    api_endpoints = ["api/v1/users", "api/v1/orders", "api/v1/login"]
    for endpoint in api_endpoints:
        api_url = urljoin(url, endpoint)
        response = send_request(api_url)
        if response and response.status_code == 200:
            results.append({
                "URL": api_url,
                "Vulnerability Type": "Exposed API Endpoint",
                "Details": f"Accessible API endpoint: {endpoint}"
            })

    # Check for Access Control Flaws
    restricted_urls = ["admin", "dashboard", "config"]
    for restricted in restricted_urls:
        restricted_url = urljoin(url, restricted)
        response = send_request(restricted_url)
        if response and response.status_code == 200:
            results.append({
                "URL": restricted_url,
                "Vulnerability Type": "Access Control Flaw",
                "Details": f"Unauthorized access to: {restricted}"
            })

    # Check for Parameter Tampering
    tamper_params = {"price": "0", "role": "admin", "id": "-1"}
    response = send_request(url)
    if response and ("admin" in response.text or "unauthorized" not in response.text):
        results.append({
            "URL": url,
            "Vulnerability Type": "Parameter Tampering",
            "Details": "Modified parameters affected server behavior."
        })

    # Check for Rate Limiting
    try:
        for i in range(10):
            response = send_request(url)
            if response and response.status_code == 200 and i == 9:
                results.append({
                    "URL": url,
                    "Vulnerability Type": "Rate Limiting",
                    "Details": "No rate limiting detected for repeated requests."
                })
    except requests.RequestException:
        pass

    # Save results to CSV
    if results:
        df = pd.DataFrame(results)
        df.to_csv(output_file, index=False)
        print(f"\nVulnerability results saved to {output_file}")
    else:
        print("\nNo vulnerabilities found.")

# Example usage
target_url = "https://www.taleblou.ir/"  # Replace with the target URL
check_url_vulnerabilities(target_url)