Task-1-Vulnerability-Scanning-Tool-OOMAP.py

import socket
import time
import re
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlparse

# Assuming `nvdlib` is mocked or replaced with a stub function for testing
def mock_nvdlib_searchCVE(cpeName):
    # Simulate a scenario where a vulnerability is found
    if "nginx" in cpeName and "1.18.0" in cpeName:
        return [{"cve": {"CVE_data_meta": {"ID": "CVE-2021-23017"}, 
                         "description": {"description_data": [{"value": "Vulnerability in nginx 1.18.0."}]}}}]
    return []

# Replace the real function with a mock for testing purposes
nvdlib = type('nvdlib', (object,), {'searchCVE': mock_nvdlib_searchCVE})

# Common weak credentials to check
weak_credentials = {
    "ftp": [("anonymous", ""), ("admin", "admin"), ("user", "password"), ("test", "test")],
    "ssh": [("root", "root"), ("admin", "admin123"), ("admin", "admin"), ("user", "password")],
    "mysql": [("root", "root"), ("admin", "admin"), ("user", "password"), ("test", "test")],
    "postgres": [("postgres", "postgres"), ("admin", "admin"), ("user", "password")],
    "telnet": [("admin", "admin"), ("user", "password"), ("root", "root")],
    "http": [("admin", "admin"), ("admin", "1234"), ("user", "password")],
    "mssql": [("sa", "password"), ("admin", "admin"), ("sa", "admin")],
    "mongodb": [("admin", "admin"), ("root", "root"), ("user", "password")],
    "rdp": [("Administrator", "admin"), ("Administrator", "password"), ("user", "password")],
    "smtp": [("admin", "admin"), ("postmaster", "postmaster"), ("root", "root")],
    "redis": [("default", ""), ("root", "root"), ("admin", "admin")],
    "vnc": [("admin", "admin"), ("root", "root"), ("user", "password")]
}

# Function to scan all ports on a target
def scan(target):
    print(f"Scanning all ports on {target}...\n")

    open_ports = []
    start_time = time.time()

    with ThreadPoolExecutor(max_workers=1000) as executor:
        futures = {executor.submit(scan_port, target, port): port for port in range(1, 65536)}

        for future in as_completed(futures):
            port = futures[future]
            if future.result():
                open_ports.append(port)

    end_time = time.time()
    print(f"\nScan completed in {end_time - start_time:.2f} seconds.")
    print("\nOpen ports:", open_ports)
    return open_ports

# Function to scan an individual port
def scan_port(target, port):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(0.5)
        result = sock.connect_ex((target, port))

        if result == 0:
            print(f"Port {port} is open on {target}")
            banner = get_banner(target, port)
            if banner:
                print(f"Service on port {port}: {banner}")
                check_for_vulnerabilities(banner)
                check_for_misconfigurations(target, port, banner)
            return True

    except socket.error:
        pass

    finally:
        sock.close()

    return False

# Function to retrieve the banner from an open port
def get_banner(target, port):
    try:
        sock = socket.socket()
        sock.settimeout(1)
        sock.connect((target, port))
        banner = sock.recv(1024).decode().strip()
        return banner
    except:
        return None
    finally:
        sock.close()

# Function to parse the banner and check for vulnerabilities using the CVE database
def check_for_vulnerabilities(banner):
    match = re.match(r"(\w+)/([\d.]+)", banner)
    if match:
        software, version = match.groups()
        print(f"Checking for vulnerabilities in {software} version {version}...")

        try:
            cpe = f"cpe:2.3:a:{software}:{software}:{version}:*:*:*:*:*:*:*"
            results = nvdlib.searchCVE(cpeName=cpe)

            if results:
                print(f"Vulnerabilities found for {software} {version}:")
                for cve in results:
                    print(f"- {cve['cve']['CVE_data_meta']['ID']}: {cve['cve']['description']['description_data'][0]['value']}")
            else:
                print(f"No known vulnerabilities found for {software} {version}.")
        except Exception as e:
            print(f"Error checking vulnerabilities: {e}")
    else:
        print("Could not parse software version from banner.")

# Function to check for common misconfigurations
# Function to check for common misconfigurations
def check_for_misconfigurations(target, port, banner):
    service = None

    # Determine the service based on the banner
    if "ftp" in banner.lower():
        service = "ftp"
    elif "ssh" in banner.lower():
        service = "ssh"
    elif "mysql" in banner.lower():
        service = "mysql"
    elif "postgres" in banner.lower():
        service = "postgres"
    elif "telnet" in banner.lower():
        service = "telnet"
    elif "http" in banner.lower():
        service = "http"
    elif "mssql" in banner.lower():
        service = "mssql"
    elif "mongodb" in banner.lower():
        service = "mongodb"
    elif "rdp" in banner.lower():
        service = "rdp"
    elif "smtp" in banner.lower():
        service = "smtp"
    elif "redis" in banner.lower():
        service = "redis"
    elif "vnc" in banner.lower():
        service = "vnc"
    
    if service:
        check_weak_credentials(target, port, service)

# Function to check weak credentials
def check_weak_credentials(target, port, service):
    if service in weak_credentials:
        for username, password in weak_credentials[service]:
            print(f"Checking {service} on port {port} with credentials {username}/{password}...")
            if try_login(target, port, service, username, password):
                print(f"Misconfiguration found: Weak credentials {username}/{password} for {service} on port {port}")
            else:
                print(f"Credentials {username}/{password} failed for {service} on port {port}")

# Mock function for login attempts (for testing purposes)
def try_login(target, port, service, username, password):
    # Placeholder logic for login attempts
    # You would replace this with actual logic using appropriate libraries for each service
    if service == "ssh" and username == "root" and password == "root":
        return True  # Simulate a successful login
    return False  # Default to unsuccessful login


# Mock function for testing login attempts
def mock_try_login(target, port, service, username, password):
    if service == "ssh" and username == "root" and password == "root":
        return True
    return False

# Replace the real try_login function with the mock one for testing purposes
try_login = mock_try_login

# Main function to run the tool
def main():
    print("Welcome To OOMap Scanner\n")
    print("Choose the Test Below for Website")
    print("1. Open Port Scanning")
    print("2. Outdated Software Versions")
    print("3. Misconfigurations")
    print("4. Test Of Outdated Software Versions & Misconfigurations")

    a = input("Enter Number for test: ")

    url = input("\nEnter the target URL: ").strip()
    parsed_url = urlparse(url)
    target = parsed_url.hostname

    if target is None:
        print("Invalid URL. Please enter a valid URL.")
        return
    
    print(f"Target domain: {target}")

    if a == "1":
        scan(target)
    
    elif a == "2":
        open_ports = scan(target)
        for port in open_ports:
            banner = get_banner(target, port)
            if banner:
                check_for_vulnerabilities(banner)
    
    elif a == "3":
        open_ports = scan(target)
        for port in open_ports:
            banner = get_banner(target, port)
            if banner:
                check_for_misconfigurations(target, port, banner)
    
    elif a == "4":
        scan(target)
        test()
    
    else:
        print("Invalid selection. Please enter 1, 2,3 or 4.")

# Test the vulnerability and misconfiguration checks with mock data
def test():
    # Test banners
    test_banners = [
        "nginx/1.18.0",  # Known vulnerability in mock function
        "apache/2.4.48",  # No vulnerabilities in mock function
        "ssh/7.6",  # Test misconfiguration for SSH
    ]

    # Test check_for_vulnerabilities
    for banner in test_banners:
        check_for_vulnerabilities(banner)

    # Test check_for_misconfigurations
    for banner in test_banners:
        check_for_misconfigurations("demo.testfire.net", 22, banner)

    # Test check_weak_credentials
    for banner in test_banners:
        service = banner.split('/')[0]
        check_weak_credentials("demo.testfire.net", 22, service)

if __name__ == "__main__":
    main()
    # Uncomment the following line to run the tests
    # test()