Python: Einfacher und flexibler Portscanner zum Schwachstellentesten (Version 0.1)

Klar man hätte auch einen der schon recht bekannten Port Scanner nutzen können, aber mich treibt es immer wieder zu einer Selbstentwicklung. Für mich ist es extrem wichtig hier Python zu nutzen, da ich es für extrem flexibel halte auch durch die Erweiterung in alle möglichen Richtungen da will ich mich schlichtweg nicht eingrenzen lassen.

Dieses Skript ist ein einfacher Port-Scanner in Python. Es scannt ein angegebenes Ziel (eine IP-Adresse oder einen Hostnamen) auf offene Ports innerhalb eines definierten Bereichs, protokolliert die Aktivitäten und erstellt einen Bericht über den Scan. Hier ist, was jeder Teil des Skripts macht:

• Konfiguration: Das Skript enthält verschiedene Konfigurationsvariablen, die entweder standardmäßig festgelegt sind oder über Umgebungsvariablen angepasst werden können. Diese Konfigurationen umfassen das Ziel des Portscans, die zu überprüfenden Portbereiche, die maximale Anzahl von Prozessen, die gleichzeitig ausgeführt werden können, die Timeout-Zeit für Socket-Operationen sowie die Namen und Pfade von Log- und Berichtsdateien.
• Logging-Konfiguration: Hier wird die Konfiguration für das Protokollieren von Ereignissen festgelegt. Es wird ein Protokollierungslevel festgelegt (standardmäßig INFO), ein Dateiname für das Protokoll (standardmäßig „port_scan.log“) und ein Format für die Protokolleinträge.
• Hilfsfunktionen: Das Skript enthält mehrere Hilfsfunktionen, die für verschiedene Aufgaben verwendet werden, z. B. das Drucken eines ASCII-Banners, das Überprüfen der Gültigkeit einer IPv6-Adresse, das Überprüfen eines bestimmten Ports, das Abrufen der IP-Adresse eines Ziels, das Generieren eines Berichts und das Generieren eines Diagramms für die Portbereiche.
• Firewall-Erkennung: Es gibt eine Funktion zum Erkennen einer Firewall oder eines IDS (Intrusion Detection System) auf dem Zielhost. Es wird eine HTTP-Anfrage an den Zielserver gesendet, und basierend auf der Antwort wird überprüft, ob eine Firewall oder ein IDS vorhanden ist.
• Portscan-Funktion: Die Hauptfunktion des Skripts ist die „scan_ports“-Funktion. Diese Funktion ruft die anderen Funktionen auf, um den Portscan durchzuführen. Sie erstellt ein ASCII-Banner, gibt Informationen zum Scan aus, erstellt eine Liste aller zu überprüfenden Ports, ruft die Funktion zum Überprüfen der Ports mithilfe von Multiprocessing auf, generiert einen Bericht, erstellt ein Diagramm für die Portbereiche und ruft die Firewall-Erkennungsfunktion auf. Der Bericht wird in einer HTML-Datei gespeichert.

Hauptprogramm: Der Code am Ende des Skripts überprüft, ob das Skript direkt ausgeführt wird (statt importiert zu werden). Wenn ja, wird die „scan_ports“-Funktion aufgerufen, um den Portscan durchzuführen.

Was in der Version 0.1 schon enthalten ist?

Hier das komplette Skript

import pyfiglet
import socket
import logging
import multiprocessing
import os
from datetime import datetime
from tqdm import tqdm
from colorama import Fore, Style
from tabulate import tabulate
import matplotlib.pyplot as plt
import requests
import ipaddress



# This line determines the target of the port scan. 
# If an environment variable "TARGET" is set, it will be used as the target. 
# If not, the default target is "example.com".
TARGET = os.environ.get("TARGET", "example.com") 


# Here, the port ranges that the scanner should check are defined.
PORT_RANGES = [
    {"start": 1, "end": 100},
    #{"start": 101, "end": 65000},
]
# This line determines the maximum number of processes that can run simultaneously. 
# By default, this is the number of CPU cores of the system, but it can be 
# customized by setting the environment variable "MAX_PROCESSES".

MAX_PROCESSES = int(os.environ.get("MAX_PROCESSES", multiprocessing.cpu_count()))

# This sets the timeout for socket operations. The default timeout is one second, 
# but it can be customized by setting the environment variable "SOCKET_TIMEOUT".
SOCKET_TIMEOUT = int(os.environ.get("SOCKET_TIMEOUT", 1))

# This determines the name of the log file. By default, it's "port_scan.log", 
# but it can be changed by setting the environment variable "LOG_FILENAME".
LOG_FILENAME = os.environ.get("LOG_FILENAME", "port_scan.log")

# This determines the path and name of the report file. By default, it's "Security_Testing/Port_Scann/Reports/port_scan_report.html", 
# but it can be changed by setting the environment variable "REPORT_FILENAME".
REPORT_FILENAME = os.environ.get("REPORT_FILENAME", "Security_Testing/Port_Scann/Reports/port_scan_report.html")

# This determines the directory where the report is saved.
REPORT_DIRECTORY = os.path.dirname(REPORT_FILENAME)
os.makedirs(REPORT_DIRECTORY, exist_ok=True) 
# This creates the directory where the report is saved, if it doesn't already exist. 
# The exist_ok=True parameter prevents an error if the directory already exists.

# Logging configuration

logging.basicConfig(filename=LOG_FILENAME, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
## filename=LOG_FILENAME: This specifies that the logs should be written to a file, 
# and the name of the file is the value of the LOG_FILENAME variable.

## level=logging.INFO: This sets the root logger level to INFO. This means that only events of this level and above will be tracked, 
# unless the logging package is configured differently elsewhere. 
# Levels you can specify include DEBUG, INFO, WARNING, ERROR, and CRITICAL.

## format='%(asctime)s - %(levelname)s - %(message)s': This is a string that specifies the layout of the log entries. 
# Each "%(name)s" is a placeholder that will be replaced with the corresponding "name" from each log record. 
# In this case, asctime will be replaced with the time the log record was created, levelname will be replaced with the level 
# name of the log record (INFO, ERROR, etc.), and message will be replaced with the log message.


def print_banner():
    ascii_banner = pyfiglet.figlet_format("OPS - Open Port Scanner")
    print(ascii_banner)

def print_scan_info():
    print("-" * 50)
    print("Scanning Target: " + TARGET)
    global start_time
    start_time = datetime.now()
    print("Scanning started at: " + str(start_time))
    print("-" * 50)

def validate_ipv6_address(ip):
    """Prüft, ob die gegebene IP-Adresse eine gültige IPv6-Adresse ist."""
    try:
        ipaddress.IPv6Address(ip)
        return True
    except ipaddress.AddressValueError:
        return False

def check_port(port_info):
    ip, port = port_info
    try:
        if validate_ipv6_address(ip):
            with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s:
                s.settimeout(SOCKET_TIMEOUT)
                # Add the scope id if it is a link-local address
                if ipaddress.IPv6Address(ip).is_link_local:
                    s.connect((ip, port, 0, YOUR_NETWORK_INTERFACE_INDEX))
                else:
                    s.connect((ip, port))
        else:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(SOCKET_TIMEOUT)
                result = s.connect_ex((ip, port))
    except OSError:
        return {"port": port, "status": "error"}

    if result == 0:
        logging.info("Port {} is open".format(port))
        return {"port": port, "status": "open"}
    else:
        return {"port": port, "status": "closed"}


def get_ip_address(target):
    addresses = socket.getaddrinfo(target, None)
    for address in addresses:
        if address[0] == socket.AF_INET6:
            # Detect if it's an IPv4-mapped IPv6 address
            ip = ipaddress.IPv6Address(address[4][0])
            if ip.ipv4_mapped:
                return str(ip.ipv4_mapped)
            else:
                return address[4][0]
    return addresses[0][4][0]


def generate_report(results):
    headers = ["Port", "Status"]
    data = [(port["port"], port["status"]) for port in results]
    table = tabulate(data, headers=headers, tablefmt="pretty")
    report = f'''
    <html>
    <head>
        <title>Port Scan Report</title>
    </head>
    <body>
        <h1>Port Scan Report</h1>
        <p>Target: {TARGET}</p>
        <p>Scan started at: {str(start_time)}</p>
        <p>Scan finished at: {str(datetime.now())}</p>
        <p>Time taken: {str(datetime.now() - start_time)}</p>
        <h2>Ports:</h2>
        {table}
    </body>
    </html>
    '''
    return report

def generate_port_range_chart(results):
    open_ports = len([port for port in results if port["status"] == "open"])
    closed_ports = len([port for port in results if port["status"] == "closed"])
    labels = ["Open Ports", "Closed Ports"]
    values = [open_ports, closed_ports]
    colors = ["green", "red"]
    explode = (0.1, 0)
    plt.pie(values, explode=explode, labels=labels, colors=colors, autopct='%1.1f%%', startangle=90)
    plt.axis('equal')
    plt.title("Port Range Chart")
    plt.show()

def detect_firewall(target):
    try:
        response = requests.get(f"http://{target}")
        if response.status_code == 200:
            print(Fore.YELLOW + "\nFirewall or IDS detected!")
            print("The response from the target server indicates the presence of a firewall or IDS.")
            print("Please exercise caution while performing the port scan.")
            print(Style.RESET_ALL)
            logging.warning("Firewall or IDS detected")
    except requests.exceptions.RequestException as e:
        print(Fore.YELLOW + "\nWarning: " + str(e))
        print(Style.RESET_ALL)
        logging.warning(str(e))
        raise e

def scan_ports():
    print_banner()
    print_scan_info()

    try:
        port_ranges = [(range_info["start"], range_info["end"]) for range_info in PORT_RANGES]
        all_ports = []

        for start, end in port_ranges:
            all_ports.extend(range(start, end + 1))

        if '/' in TARGET:
            if ':' in TARGET:
                ip_network = ipaddress.IPv6Network(TARGET, strict=False)
            else:
                ip_network = ipaddress.IPv4Network(TARGET, strict=False)
            target_ips = [str(ip) for ip in ip_network.hosts()]
        elif ',' in TARGET:
            target_ips = [ip.strip() for ip in TARGET.split(',')]
        else:
            target_ips = [socket.getaddrinfo(TARGET, None, socket.AF_UNSPEC)[0][4][0]]
        '''Finally, you will need to adjust your code to handle IPv6 link-local addresses correctly when you are resolving the TARGET:
            In this line, I have changed socket.AF_UNSPEC to socket.AF_INET6 to ensure that it will resolve to an IPv6 address.'''    
        #else:
            #target_ips = [socket.getaddrinfo(TARGET, None, socket.AF_INET6)[0][4][0]]    

        with multiprocessing.Pool(processes=MAX_PROCESSES) as pool:
            total_ports = len(all_ports) * len(target_ips)
            results = list(tqdm(pool.imap(check_port, [(ip, port) for ip in target_ips for port in all_ports]), total=total_ports, ncols=80, bar_format='{l_bar}{bar}| {n_fmt}/{total_fmt}'))

        open_ports = [port["port"] for port in results if port["status"] == "open"]

        if open_ports:
            print(Fore.GREEN + "\nOpen Ports:")
            for port in open_ports:
                print("Port {} is open".format(port))
            print(Style.RESET_ALL)
            logging.info("Open ports: {}".format(open_ports))
        else:
            print(Fore.RED + "\nNo open ports found." + Style.RESET_ALL)
            logging.info("No open ports found.")

        report = generate_report(results)

        with open(REPORT_FILENAME, 'w') as report_file:
            report_file.write(report)

        generate_port_range_chart(results)

        detect_firewall(TARGET)

        print("Port scan report generated: {}".format(REPORT_FILENAME))

    except KeyboardInterrupt:
        print("\nExiting Program !!!!")
        logging.info("Program interrupted by the user")
    except socket.error as e:
        print("\nServer not responding !!!!")
        print(Fore.RED + "\nError: " + str(e))
        logging.error(str(e))
        raise e
    except requests.exceptions.RequestException as e:
        print(Fore.YELLOW + "\nWarning: " + str(e))
        logging.warning(str(e))
        raise e

if __name__ == '__main__':
    scan_ports()