Upload files to ''

createTargers.py

@@ -0,0 +1,154 @@
+import csv
+from gvm.connections import TLSConnection
+from gvm.protocols.gmp import Gmp
+from gvm.transforms import EtreeTransform
+
+
+# OpenVAS Configuration
+OPENVAS_HOST = "localhost"
+OPENVAS_PORT = 9390
+USERNAME = "admin"
+PASSWORD = "admin"
+
+# CSV files
+ACTIVE_HOSTS_CSV = "active_hosts.csv"
+TARGET_ID_CSV = "target_id.csv"
+
+
+def get_port_list_id(gmp):
+    """
+    Retrieve a valid port list ID.
+    """
+    port_lists = gmp.get_port_lists()
+    default_port_list_id = None
+    for port_list in port_lists.findall("port_list"):
+        name = port_list.find("name").text
+        port_list_id = port_list.get("id")
+        print(f"Port List: {name} (ID: {port_list_id})")
+        if "OpenVAS Default" in name:
+            default_port_list_id = port_list_id
+
+    # If "OpenVAS Default" is not found, use the first available port list
+    if not default_port_list_id and port_lists:
+        default_port_list_id = port_lists.find("port_list").get("id")
+        print(f"'OpenVAS Default' not found. Using the first available port list with ID: {default_port_list_id}")
+
+    return default_port_list_id
+
+
+def target_exists(gmp, ip):
+    """
+    Check if a target already exists for the given IP and return its target_id if it does.
+    """
+    targets = gmp.get_targets()
+    for target in targets.findall("target"):
+        if target.find("hosts").text == ip:
+            target_id = target.get("id")
+            print(f"Target for IP {ip} already exists with ID: {target_id}")
+            return target_id
+    return None
+
+
+def create_target(gmp, name, ip, port_list_id):
+    """
+    Create a target in OpenVAS.
+    """
+    response = gmp.create_target(name=name, hosts=ip, port_list_id=port_list_id)
+    target_id = response.get("id")
+    if target_id:
+        print(f"Created target '{name}' for IP '{ip}' with ID: {target_id}")
+    else:
+        print(f"Failed to create target for IP {ip}. Check logs.")
+    return target_id
+
+
+def save_target_id_to_csv(ip, target_id):
+    """
+    Save the target ID to the target_id.csv file if it's not already there.
+    """
+    existing_data = []
+    try:
+        # Read existing data from the CSV
+        with open(TARGET_ID_CSV, "r", newline="") as csvfile:
+            reader = csv.reader(csvfile)
+            existing_data = list(reader)
+    except FileNotFoundError:
+        # If the file doesn't exist, it will be created later
+        pass
+
+    # Check if the IP is already in the CSV
+    for row in existing_data:
+        if row[0] == ip:
+            print(f"IP {ip} already recorded in {TARGET_ID_CSV}")
+            return
+
+    # Append the new target ID to the CSV
+    with open(TARGET_ID_CSV, "a", newline="") as csvfile:
+        writer = csv.writer(csvfile)
+        writer.writerow([ip, target_id])
+    print(f"Saved target ID {target_id} for IP {ip} to {TARGET_ID_CSV}")
+
+
+def read_csv_to_ip_list(csv_file):
+    """
+    Read the CSV file and extract IPs into a list.
+    """
+    ip_list = []
+    with open(csv_file, newline="") as csvfile:
+        reader = csv.DictReader(csvfile)
+        for row in reader:
+            ip_list.append(row["IP"])
+    return ip_list
+
+
+def main():
+    connection = TLSConnection(hostname=OPENVAS_HOST, port=OPENVAS_PORT)
+    with Gmp(connection=connection, transform=EtreeTransform()) as gmp:
+        # Authenticate with OpenVAS
+        gmp.authenticate(username=USERNAME, password=PASSWORD)
+        print("Authenticated with OpenVAS")
+
+        # Get a valid port list ID
+        port_list_id = get_port_list_id(gmp)
+        if not port_list_id:
+            print("Failed to retrieve a valid port list. Exiting.")
+            return
+
+        # Read IPs from the active hosts CSV
+        ip_list = read_csv_to_ip_list(ACTIVE_HOSTS_CSV)
+        if not ip_list:
+            print("No IPs found in the CSV file. Exiting.")
+            return
+
+        print(f"Found {len(ip_list)} active hosts.")
+
+        # Process each IP
+        for ip in ip_list:
+            print(f"Processing IP: {ip}")
+
+            # Check if target already exists
+            target_id = target_exists(gmp, ip)
+            if target_id:
+                # Save the existing target ID to the CSV if not already recorded
+                save_target_id_to_csv(ip, target_id)
+                continue
+
+            # Create target if it doesn't exist
+            target_name = f"Target for {ip}"
+            target_id = create_target(gmp, target_name, [ip], port_list_id)
+            if target_id:
+                save_target_id_to_csv(ip, target_id)
+            else:
+                print(f"Failed to create target for IP {ip}. Continuing.")
+
+
+if __name__ == "__main__":
+    # Ensure the target_id.csv file exists with headers
+    try:
+        with open(TARGET_ID_CSV, "x", newline="") as csvfile:
+            writer = csv.writer(csvfile)
+            writer.writerow(["IP", "Target ID"])  # Write headers if the file doesn't exist
+    except FileExistsError:
+        pass  # File already exists, no need to create
+
+    main()

networkdiscovery.py

@@ -0,0 +1,41 @@
+import nmap
+import psutil
+import socket
+import ipaddress
+import pandas as pd
+from datetime import datetime
+
+def get_ipv4_interfaces():
+    ipv4_interfaces = []
+    for name, addrs in psutil.net_if_addrs().items():
+        for addr in addrs:
+            if addr.family == socket.AF_INET:
+                ipv4_interfaces.append((name, addr.address, addr.netmask))
+    return ipv4_interfaces
+
+def network_discovery():
+    ipv4_interfaces = get_ipv4_interfaces()
+    if not ipv4_interfaces:
+        print("No active IPv4 network interfaces found.")
+        return
+
+    nm = nmap.PortScanner()
+    active_hosts = []
+
+    for name, ip, netmask in ipv4_interfaces:
+        network = ipaddress.IPv4Network((ip, netmask), strict=False)
+        if str(network) == "127.0.0.0/8":
+            continue
+
+        print(f"Discovering hosts in IP range {network}...")
+        nm.scan(hosts=str(network), arguments='-sn -T4 --open')
+        
+        for host in nm.all_hosts():
+            active_hosts.append([host, datetime.now().strftime("%Y-%m-%d %H:%M:%S")])
+
+    columns = ['IP', 'Discovery Timestamp']
+    df = pd.DataFrame(active_hosts, columns=columns)
+    df.to_csv('active_hosts.csv', index=False)
+
+if __name__ == "__main__":
+    network_discovery()

portscanner.py

@@ -0,0 +1,54 @@
+import nmap
+import pandas as pd
+from datetime import datetime
+import os
+
+def detailed_scan():
+    try:
+        active_hosts_df = pd.read_csv('active_hosts.csv')
+    except FileNotFoundError:
+        print("File active_hosts.csv not found. Please run the network discovery script first.")
+        return
+
+    nm = nmap.PortScanner()
+    scan_results = []
+
+    start_time = datetime.now()
+    print(f"Scan started at: {start_time.strftime('%Y-%m-%d %H:%M:%S')}")
+
+    for index, row in active_hosts_df.iterrows():
+        ip = row['IP']
+        print(f"Scanning {ip} for detailed information...")
+        nm.scan(ip, arguments='-p- -T5 ')
+
+        for host in nm.all_hosts():
+            hostnames = [x['name'] for x in nm[host].get('hostnames', [])]
+            mac_address = nm[host]['addresses'].get('mac', 'N/A')
+            scan_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+            for proto in nm[host].all_protocols():
+                for port in nm[host][proto]:
+                    state = nm[host][proto][port]['state']
+                    name = nm[host][proto][port]['name']
+                    product = nm[host][proto][port]['product']
+                    version = nm[host][proto][port]['version']
+                    extrainfo = nm[host][proto][port]['extrainfo']
+                    row = [ip, ', '.join(hostnames), mac_address, proto, port, name, state, product, version, extrainfo, scan_timestamp]
+                    scan_results.append(row)
+
+    end_time = datetime.now()
+    print(f"Scan finished at: {end_time.strftime('%Y-%m-%d %H:%M:%S')}")
+    print(f"Total scan duration: {end_time - start_time}")
+
+    columns = ['IP', 'Hostname', 'MAC Address', 'Protocol', 'Port', 'Name', 'State', 'Product', 'Version', 'Extra Info', 'Timestamp']
+    detailed_df = pd.DataFrame(scan_results, columns=columns)
+
+    if os.path.exists('detailed_scan_results.csv'):
+        detailed_df.to_csv('detailed_scan_results.csv', mode='a', header=False, index=False)
+    else:
+        detailed_df.to_csv('detailed_scan_results.csv', index=False)
+
+    print(detailed_df)
+
+if __name__ == "__main__":
+    detailed_scan()

starttask.py

@@ -0,0 +1,62 @@
+import csv
+from gvm.connections import TLSConnection
+from gvm.protocols.gmp import Gmp
+from gvm.transforms import EtreeTransform
+
+
+# OpenVAS Configuration
+OPENVAS_HOST = "localhost"
+OPENVAS_PORT = 9390
+USERNAME = "admin"
+PASSWORD = "admin"
+
+# CSV file
+TASK_ID_CSV = "task_id.csv"
+
+
+def start_task(gmp, task_id):
+    """
+    Start a task in OpenVAS.
+    """
+    try:
+        gmp.start_task(task_id=task_id)
+        print(f"Started task with ID: {task_id}")
+    except Exception as e:
+        print(f"Failed to start task with ID {task_id}. Error: {e}")
+
+
+def read_csv_to_task_list(csv_file):
+    """
+    Read the CSV file and extract task IDs into a list.
+    """
+    task_list = []
+    with open(csv_file, newline="") as csvfile:
+        reader = csv.DictReader(csvfile)
+        for row in reader:
+            task_list.append(row["Task ID"])
+    return task_list
+
+
+def main():
+    connection = TLSConnection(hostname=OPENVAS_HOST, port=OPENVAS_PORT)
+    with Gmp(connection=connection, transform=EtreeTransform()) as gmp:
+        # Authenticate with OpenVAS
+        gmp.authenticate(username=USERNAME, password=PASSWORD)
+        print("Authenticated with OpenVAS")
+
+        # Read task IDs from the task_id.csv file
+        task_list = read_csv_to_task_list(TASK_ID_CSV)
+        if not task_list:
+            print("No task IDs found in the CSV file. Exiting.")
+            return
+
+        print(f"Found {len(task_list)} tasks to start.")
+
+        # Start each task
+        for task_id in task_list:
+            start_task(gmp, task_id)
+
+
+if __name__ == "__main__":
+    main()
+

taskmaker.py

@@ -0,0 +1,135 @@
+import csv
+from gvm.connections import TLSConnection
+from gvm.protocols.gmp import Gmp
+from gvm.transforms import EtreeTransform
+
+
+# OpenVAS Configuration
+OPENVAS_HOST = "localhost"
+OPENVAS_PORT = 9390
+USERNAME = "admin"
+PASSWORD = "admin"
+
+# CSV files
+TARGET_ID_CSV = "target_id.csv"
+TASK_ID_CSV = "task_id.csv"
+
+
+def get_scan_config_id(gmp):
+    """
+    Retrieve the ID of the 'Full and Fast' scan configuration.
+    """
+    scan_configs = gmp.get_scan_configs()
+    for config in scan_configs.findall("config"):
+        if config.find("name").text == "Full and fast":
+            config_id = config.get("id")
+            print(f"'Full and Fast' scan configuration found with ID: {config_id}")
+            return config_id
+    print("Failed to find 'Full and Fast' scan configuration.")
+    return None
+
+
+def get_scanner_id(gmp):
+    """
+    Retrieve the ID of the default scanner.
+    """
+    scanners = gmp.get_scanners()
+    for scanner in scanners.findall("scanner"):
+        if "OpenVAS Default" in scanner.find("name").text:
+            scanner_id = scanner.get("id")
+            print(f"'OpenVAS Default' scanner found with ID: {scanner_id}")
+            return scanner_id
+    print("Failed to find 'OpenVAS Default' scanner.")
+    return None
+
+
+
+def read_csv_to_target_list(csv_file):
+    """
+    Read the CSV file and extract target IDs into a list.
+    """
+    target_list = []
+    with open(csv_file, newline="") as csvfile:
+        reader = csv.DictReader(csvfile)
+        for row in reader:
+            target_list.append(row["Target ID"])
+    return target_list
+
+def save_task_id_to_csv(target_id, task_id):
+    """
+    Save the task ID to the task_id.csv file.
+    """
+    with open(TASK_ID_CSV, "a", newline="") as csvfile:
+        writer = csv.writer(csvfile)
+        writer.writerow([target_id, task_id])
+    print(f"Saved task ID {task_id} for target ID {target_id} to {TASK_ID_CSV}")
+
+
+def create_task(gmp, task_name, target_id, scan_config_id, scanner_id):
+    """
+    Create a task in OpenVAS.
+    """
+    response = gmp.create_task(
+        name=task_name,
+        config_id=scan_config_id,
+        target_id=target_id,
+        scanner_id=scanner_id
+    )
+    task_id = response.get("id")
+    if task_id:
+        print(f"Created task '{task_name}' for target ID '{target_id}' with ID: {task_id}")
+    else:
+        print(f"Failed to create task for target ID {target_id}. Check logs.")
+    return task_id
+
+
+def main():
+    connection = TLSConnection(hostname=OPENVAS_HOST, port=OPENVAS_PORT)
+    with Gmp(connection=connection, transform=EtreeTransform()) as gmp:
+        # Authenticate with OpenVAS
+        gmp.authenticate(username=USERNAME, password=PASSWORD)
+        print("Authenticated with OpenVAS")
+
+        # Get the scan configuration ID
+        scan_config_id = get_scan_config_id(gmp)
+        if not scan_config_id:
+            print("Failed to retrieve a valid scan configuration. Exiting.")
+            return
+
+        # Get the scanner ID
+        scanner_id = get_scanner_id(gmp)
+        if not scanner_id:
+            print("Failed to retrieve a valid scanner. Exiting.")
+            return
+
+        # Read target IDs from the target_id.csv file
+        target_list = read_csv_to_target_list(TARGET_ID_CSV)
+        if not target_list:
+            print("No target IDs found in the CSV file. Exiting.")
+            return
+
+        print(f"Found {len(target_list)} targets to create tasks for.")
+
+        # Process each target
+        for target_id in target_list:
+            print(f"Processing target ID: {target_id}")
+
+            # Create task for the target
+            task_name = f"Task for Target {target_id}"
+            task_id = create_task(gmp, task_name, target_id, scan_config_id, scanner_id)
+            if task_id:
+                save_task_id_to_csv(target_id, task_id)
+            else:
+                print(f"Failed to create task for target ID {target_id}. Continuing.")
+
+
+if __name__ == "__main__":
+    # Ensure the task_id.csv file exists with headers
+    try:
+        with open(TASK_ID_CSV, "x", newline="") as csvfile:
+            writer = csv.writer(csvfile)
+            writer.writerow(["Target ID", "Task ID"])  # Write headers if the file doesn't exist
+    except FileExistsError:
+        pass  # File already exists, no need to create
+
+    main()