create-tasks-from-csv.gmp.py

# SPDX-FileCopyrightText: 2024 Martin Boller
#
# SPDX-License-Identifier: GPL-3.0-or-later
#

# Loosely based on the create-targets-from-host-list
#
# Run with gvm-script --gmp-username admin-user --gmp-password password socket create-tasks-from-csv.gmp.py hostname-server tasks.csv

# Note: for some weird reason theres a space in front of the default scan config " Full and fast"
# Examples of all defaults in tasks.csv

import csv
import sys
import time
from argparse import ArgumentParser, Namespace, RawTextHelpFormatter
from pathlib import Path

from gvm.errors import GvmResponseError
from gvm.protocols.gmp import Gmp
from gvmtools.helper import error_and_exit

HELP_TEXT = (
    "This script pulls taskname, scanner, & scan config "
    "from a csv file and creates a task for each row. \n\n"
    "csv file to contain name of task, target name, scanner name, and scan config name  \n"
    "name,target,scanner,scan config"
)


def check_args(args):
    len_args = len(args.script) - 1
    if len_args != 1:
        message = """
        This script pulls taskinformation from a csv file and creates a task \
for each row.
        One parameter after the script name is required.

        1. <tasks_csvfile>  -- text file containing taskname, target name, scanner name, and scan config name

        Example:
            $ gvm-script --gmp-username name --gmp-password pass \
ssh --hostname <gsm> scripts/create_tasks_from_csv.gmp <tasks_csvfile>
        """
        print(message)
        sys.exit()


def parse_args(args: Namespace) -> Namespace:  # pylint: disable=unused-argument
    """Parsing args ..."""

    parser = ArgumentParser(
        prefix_chars="+",
        add_help=False,
        formatter_class=RawTextHelpFormatter,
        description=HELP_TEXT,
    )

    parser.add_argument(
        "+h",
        "++help",
        action="help",
        help="Show this help message and exit.",
    )

    parser.add_argument(
        "tasks_csv_file",
        type=str,
        help=("File containing host names / IPs"),
    )

    ports = parser.add_mutually_exclusive_group()
    ports.add_argument(
        "+pl",
        "++port-list-id",
        type=str,
        dest="port_list_id",
        help="UUID of existing port list.",
    )
    ports.add_argument(
        "+pr",
        "++port-range",
        dest="port_range",
        type=str,
        help=(
            "Port range to create port list from, e.g. "
            "T:1-1234 for ports 1-1234/TCP"
        ),
    )

    ports.set_defaults(
        port_list_id="730ef368-57e2-11e1-a90f-406186ea4fc5"
    )  # All IANA assigned TCP and Top 100 UDP, use the list-portlists.gmp.py script to get the UUID for all.
    # Defaults are
    # # | Name                          | ID
    # - | ----------------------------- | ------------------------------------
    # 1 | All IANA assigned TCP         | 33d0cd82-57c6-11e1-8ed1-406186ea4fc5
    # 2 | All IANA assigned TCP and UDP | 4a4717fe-57d2-11e1-9a26-406186ea4fc5
    # 3 | All TCP and Nmap top 100 UDP  | 730ef368-57e2-11e1-a90f-406186ea4fc5

    script_args, _ = parser.parse_known_args(args)
    return script_args


def config_id(
    gmp: Gmp,
    config_name: str,
):
    response_xml = gmp.get_scan_configs(
        filter_string="rows=-1, name= " + config_name
    )
    scan_configs_xml = response_xml.xpath("config")
    config_id = ""

    for scan_config in scan_configs_xml:
        config_id = scan_config.get("id")
    return config_id


def alert_id(
    gmp: Gmp,
    alert_name: str,
):
    response_xml = gmp.get_alerts(filter_string="rows=-1, name=" + alert_name)
    alerts_xml = response_xml.xpath("alert")
    alert_id = ""

    for alert in alerts_xml:
        alert_id = alert.get("id")
    return alert_id


def target_id(
    gmp: Gmp,
    target_name: str,
):
    response_xml = gmp.get_targets(filter_string="rows=-1, name=" + target_name)
    targets_xml = response_xml.xpath("target")
    target_id = ""

    for target in targets_xml:
        target_id = target.get("id")
    return target_id


def scanner_id(
    gmp: Gmp,
    scanner_name: str,
):
    response_xml = gmp.get_scanners(
        filter_string="rows=-1, name=" + scanner_name
    )
    scanners_xml = response_xml.xpath("scanner")
    scanner_id = ""

    for scanner in scanners_xml:
        scanner_id = scanner.get("id")
    return scanner_id


def schedule_id(
    gmp: Gmp,
    schedule_name: str,
):
    response_xml = gmp.get_schedules(
        filter_string="rows=-1, name=" + schedule_name
    )
    schedules_xml = response_xml.xpath("schedule")
    schedule_id = ""

    for schedule in schedules_xml:
        schedule_id = schedule.get("id")
    return schedule_id


def task_id(
    gmp: Gmp,
    taskName: str,
):
    response_xml = gmp.get_tasks(filter_string="rows=-1, name=" + taskName)
    tasks_xml = response_xml.xpath("task")
    task_id = ""

    for task in tasks_xml:
        task_id = task.get("id")
    return task_id


def create_tasks(
    gmp: Gmp,
    task_csv_file: Path,
    port_list_id: str,
):
    try:
        numberTasks = 0
        with open(task_csv_file, encoding="utf-8") as csvFile:
            content = csv.reader(csvFile, delimiter=",")  # read the data
            for row in content:  # loop through each row
                if len(row) == 0:
                    continue
                name = row[0]
                targetId = target_id(gmp, row[1])
                scannerId = scanner_id(gmp, row[2])
                alterable = "True"
                configId = config_id(gmp, row[3])
                scheduleId = schedule_id(gmp, row[4])

                newOrder = row[5].upper()
                if newOrder == "RANDOM":
                    order = gmp.types.HostsOrdering.RANDOM
                elif newOrder == "SEQUENTIAL":
                    order = gmp.types.HostsOrdering.SEQUENTIAL
                elif newOrder == "REVERSE":
                    order = gmp.types.HostsOrdering.REVERSE
                else:
                    order = gmp.types.HostsOrdering.RANDOM

                alerts = []
                if len(row[6]) > 1:
                    alert = alert_id(gmp, row[6])
                    alerts.append(alert)
                if len(row[7]) > 1:
                    alert = alert_id(gmp, row[7])
                    alerts.append(alert)
                if len(row[8]) > 1:
                    alert = alert_id(gmp, row[8])
                    alerts.append(alert)
                if len(row[9]) > 1:
                    alert = alert_id(gmp, row[9])
                    alerts.append(alert)
                if len(row[10]) > 1:
                    alert = alert_id(gmp, row[10])
                    alerts.append(alert)

                scanOrder = order  # Use SEQUENTIAL, REVERSE, or RANDOM
                comment = f"Created: {time.strftime('%Y/%m/%d-%H:%M:%S')}"

                try:
                    if task_id(gmp, name):
                        print(f"Task: {name} exist already")
                        continue
                    print("Creating task: " + name)
                    gmp.create_task(
                        name=name,
                        comment=comment,
                        config_id=configId,
                        target_id=targetId,
                        hosts_ordering=scanOrder,
                        scanner_id=scannerId,
                        alterable=alterable,
                        schedule_id=scheduleId,
                        alert_ids=alerts,
                    )
                    numberTasks = numberTasks + 1
                except GvmResponseError as gvmerr:
                    print(f"{gvmerr=}, name: {name}")
                    pass
        csvFile.close()  # close the csv file

    except IOError as e:
        error_and_exit(f"Failed to read task_csv_file: {str(e)} (exit)")

    if len(row) == 0:
        error_and_exit("Host file is empty (exit)")

    return numberTasks


def main(gmp: Gmp, args: Namespace) -> None:
    # pylint: disable=undefined-variable
    if args.script:
        args = args.script[1:]

    parsed_args = parse_args(args=args)
    # port_list_id="4a4717fe-57d2-11e1-9a26-406186ea4fc5"

    print("Creating tasks\n")

    numberTasks = create_tasks(
        gmp, parsed_args.tasks_csv_file, parsed_args.port_list_id
    )

    numberTasks = str(numberTasks)
    print("    [" + numberTasks + "] task(s) created!\n")


if __name__ == "__gmp__":
    main(gmp, args)