start-alert-scan.gmp.main() - Code Metrics - Inspection of "Increase coverage" - greenbone/gvm-tools - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#344)

by Jaspar

created 2020-12-08 21:01 UTC

start-alert-scan.gmp.main() B

↳ Parent: start-alert-scan.gmp

Complexity

Conditions

Size

Total Lines	129
Code Lines	89

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	89
nop	2
dl	0
loc	129
rs	7.36
c	0
b	0
f	0

How to fix Long Method

# -*- coding: utf-8 -*-
# Copyright (C) 2018 Henning Häcker
#
# SPDX-License-Identifier: GPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import sys
from typing import List
from argparse import ArgumentParser, RawTextHelpFormatter


HELP_TEXT = """
        This script makes an E-Mail alert scan.

        Usage examples: 
            $ gvm-script --gmp-username name --gmp-password pass ssh --hostname
            ... start-alert-scan.gmp.py +h
            ... start-alert-scan.gmp.py ++target-name ++hosts ++ports \
                    ++port-list-name +C ++recipient ++sender
            ... start-alert-scan.gmp.py ++target-name ++hosts ++port-list-id \
                    +C ++recipient ++sender
            ... start-alert-scan.gmp.py ++target-id +C ++recipient ++sender
    """


def get_config(gmp, config, debug=False):
    # get all configs of the openvas instance
    # filter for all rows!
    res = gmp.get_configs(filter="rows=-1")

    if config < 0 or config > 4:
        raise ValueError("Wrong config identifier. Choose between [0,4].")
    # match the config abbreviation to accepted config names
    config_list = [
        'Full and fast',
        'Full and fast ultimate',
        'Full and very deep',
        'Full and very deep ultimate',
        'System Discovery',
    ]
    template_abbreviation_mapper = {
        0: config_list[0],
        1: config_list[1],
        2: config_list[2],
        3: config_list[3],
        4: config_list[4],
    }

    config_id = "-"
    for conf in res.xpath('config'):
        cid = conf.xpath('@id')[0]
        name = conf.xpath('name/text()')[0]

        # get the config id of the desired template
        if template_abbreviation_mapper.get(config, "-") == name:
            config_id = cid
            if debug:
                print(name + ": " + config_id)
            break

    # check for existence of the desired config
    if config_id == "-":
        print(
            "error: could not recognize template '%s'"
            "\nvalid template names are: %s\n" % (template, config_list)

        )
        exit()

    return config_id


def get_target(
    gmp,
    target_name: str = None,
    hosts: List[str] = None,
    ports: str = None,
    port_list_name: str = None,
    port_list_id: str = None,
    debug: bool = False,
):
    targets = gmp.get_targets()
    counter = 0
    exists = True
    # iterate over existing targets and find a vacant targetName
    while exists:
        exists = False
        if target_name is None:
            target_name = "target_{}".format(str(counter))
        for target in targets.xpath('target'):
            name = target.xpath('name/text()')[0]
            if name == target_name:
                exists = True
                break
        counter += 1

    if debug:
        print("target name: {}".format(target_name))

    if not port_list_id:
        # iterate over existing port lists and find a vacant name
        existing_port_lists = [""]
        for name in gmp.get_port_lists().findall("port_list/name"):
            existing_port_lists.append(str(name.text))

        if port_list_name is None:
            port_list_name = "portlist_{}".format(str(counter))
        counter = 0
        while True:
            if port_list_name not in existing_port_lists:
                break
            counter += 1

        # create port list
        portlist = gmp.create_port_list(portlist_name, ports)

        port_list_id = portlist.xpath('@id')[0]
        if debug:
            print("New Portlist-name:\t{}".format(str(portlist_name)))
            print("New Portlist-id:\t{}".format(str(portlist_id)))


    # integrate port list id into create_target
    res = gmp.create_target(target_name, hosts=hosts, port_list_id=port_list_id)
    return res.xpath('@id')[0]


def get_alert(
    gmp,
    sender_email: str,
    recipient_email: str,
    alert_name: str = None,
    debug=False,
):

    # create alert if necessary
    alert_object = gmp.get_alerts(filter='name={}'.format(alert_name))
    alert = alert_object.xpath('alert')

    if len(alert) == 0:
        print("creating new alert {}".format(alert_name))
        gmp.create_alert(
            alert_name,
            event=gmp.types.AlertEvent.TASK_RUN_STATUS_CHANGED,
            event_data={"status": "Done"},
            condition=gmp.types.AlertCondition.ALWAYS,
            method=gmp.types.AlertMethod.EMAIL,
            method_data={
                """Task '$n': $e

After the event $e,
the following condition was met: $c

This email escalation is configured to attach report format '$r'.
Full details and other report formats are available on the scan engine.

$t

Note:
This email was sent to you as a configured security scan escalation.
Please contact your local system administrator if you think you
should not have received it.
""": "message",
                "2": "notice",
                sender_email: "from_address",
                "[OpenVAS-Manager] Task": "subject",
                "c402cc3e-b531-11e1-9163-406186ea4fc5": "notice_attach_format",
                recipient_email: "to_address",
            },
        )

        alert_object = gmp.get_alerts(filter='name={}'.format(recipient_email))
        alert = alert_object.xpath('alert')

    alert_id = alert[0].get('id', 'no id found')
    if debug:
        print("alert_id: {}".format(str(alert_id)))

    return alert_id


def get_scanner(gmp):
    res = gmp.get_scanners()
    scanner_ids = res.xpath('scanner/@id')
    return scanner_ids[1]  # default scanner


def create_and_start_task(
    gmp,
    config_id: str,
    target_id: str,
    scanner_id: str,
    alert_id: str,
    alert_name: str,
    debug: bool = False,
):
    # Create the task
    tasks = gmp.get_tasks(filter="name~{}".format(alert_name))
    task_name = "Alert Scan for Alert {}".format(alert_name)
    if tasks:
        task_name = "Alert Scan for Alert {} ({})".format(
            alert_name, len(tasks.xpath('tasks/@id'))
        )
    task_comment = "Alert Scan"
    res = gmp.create_task(
        task_name,
        config_id,
        target_id,
        scanner_id,
        alert_ids=[alert_id],
        comment=task_comment,
    )

    # Start the task
    task_id = res.xpath('@id')[0]
    gmp.start_task(task_id)

    print('Task started: ' + task_name)

    if debug:
        # Stop the task (for performance reasons)
        gmp.stop_task(task_id)
        print('Task stopped')


def main(gmp, args):
    # pylint: disable=undefined-variable

    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.",
    )

    target = parser.add_mutually_exclusive_group(required=True)

    target.add_argument(
        "++target-id",
        type=str,
        dest="target_id",
        help="Use an existing by target id",
    )

    target.add_argument(
        "++target-name",
        type=str,
        dest="target_name",
        help="Create a target by name",
    )

    parser.add_argument(
        "++hosts",
        nargs='+',
        dest='hosts',
        help="Host(s) for the new target",
    )

    ports = parser.add_mutually_exclusive_group()

    ports.add_argument(
        "++port-list-id",
        type=str,
        dest="port_list_id",
        help="An existing portlist id for the new target",
    )
    ports.add_argument(
        "++ports",
        type=str,
        dest='ports',
        help="Ports in the new target: e.g. T:80-80,8080",
    )

    parser.add_argument(
        "++port-list-name",
        type=str,
        dest="port_list_name",
        help="Name for the new portlist in the new target",
    )

    parser.add_argument(
        "+C",
        "++scan_config",
        default=0,
        type=int,
        dest='config',
        help="Choose from existing scan config:"
        " 0: Full and fast"
        " 1: Full and fast ultimate"
        " 2: Full and very deep"
        " 3: Full and very deep ultimate"
        " 4: System Discovery",
    )

    parser.add_argument(
        "++recipient",
        required=True,
        dest='recipient_email',
        type=str,
        help="Alert recipient E-Mail address",
    )

    parser.add_argument(
        "++sender",
        required=True,
        dest='sender_email',
        type=str,
        help="Alert senders E-Mail address",
    )

    parser.add_argument(
        "++alert-name",
        dest='alert_name',
        type=str,
        help="Optional Alert name",
    )

    script_args, _ = parser.parse_known_args()

    if script_args.alert_name is None:
        script_args.alert_name = script_args.recipient_email

    config_id = get_config(gmp, script_args.config)
    if not script_args.target_id:
        target_id = get_target(
            gmp,
            target_name=script_args.target_id,
            hosts=script_args.hosts,
            ports=script_args.ports,
            port_list_name=script_args.port_list_name,
            port_list_id=script_args.port_list_id,
        )
    else:
        target_id = script_args.target_id
    alert_id = get_alert(
        gmp,
        script_args.sender_email,
        script_args.recipient_email,
        script_args.alert_name,
    )
    scanner_id = get_scanner(gmp)

    create_and_start_task(
        gmp, config_id, target_id, scanner_id, alert_id, script_args.alert_name
    )

    print("\nScript finished\n")


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



1			# -- coding: utf-8 --
2			# Copyright (C) 2018 Henning Häcker
3			#
4			# SPDX-License-Identifier: GPL-3.0-or-later
5			#
6			# This program is free software: you can redistribute it and/or modify
7			# it under the terms of the GNU General Public License as published by
8			# the Free Software Foundation, either version 3 of the License, or
9			# (at your option) any later version.
10			#
11			# This program is distributed in the hope that it will be useful,
12			# but WITHOUT ANY WARRANTY; without even the implied warranty of
13			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14			# GNU General Public License for more details.
15			#
16			# You should have received a copy of the GNU General Public License
17			# along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19			import sys
20			from typing import List
21			from argparse import ArgumentParser, RawTextHelpFormatter
22
23
24			HELP_TEXT = """
25			This script makes an E-Mail alert scan.
26
27			Usage examples:
28			$ gvm-script --gmp-username name --gmp-password pass ssh --hostname
29			... start-alert-scan.gmp.py +h
30			... start-alert-scan.gmp.py ++target-name ++hosts ++ports \
31			++port-list-name +C ++recipient ++sender
32			... start-alert-scan.gmp.py ++target-name ++hosts ++port-list-id \
33			+C ++recipient ++sender
34			... start-alert-scan.gmp.py ++target-id +C ++recipient ++sender
35			"""
36
37
38			def get_config(gmp, config, debug=False):
39			# get all configs of the openvas instance
40			# filter for all rows!
41			res = gmp.get_configs(filter="rows=-1")
42
43			if config < 0 or config > 4:
44			raise ValueError("Wrong config identifier. Choose between [0,4].")
45			# match the config abbreviation to accepted config names
46			config_list = [
47			'Full and fast',
48			'Full and fast ultimate',
49			'Full and very deep',
50			'Full and very deep ultimate',
51			'System Discovery',
52			]
53			template_abbreviation_mapper = {
54			0: config_list[0],
55			1: config_list[1],
56			2: config_list[2],
57			3: config_list[3],
58			4: config_list[4],
59			}
60
61			config_id = "-"
62			for conf in res.xpath('config'):
63			cid = conf.xpath('@id')[0]
64			name = conf.xpath('name/text()')[0]
65
66			# get the config id of the desired template
67			if template_abbreviation_mapper.get(config, "-") == name:
68			config_id = cid
69			if debug:
70			print(name + ": " + config_id)
71			break
72
73			# check for existence of the desired config
74			if config_id == "-":
75			print(
76			"error: could not recognize template '%s'"
77			"\nvalid template names are: %s\n" % (template, config_list)
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-12-07 08:53 UTC by Report Bug Copy Issue Report The variable `template` does not seem to be defined. Loading history...
78			)
79			exit()
80
81			return config_id
82
83
84			def get_target(
85			gmp,
86			target_name: str = None,
87			hosts: List[str] = None,
88			ports: str = None,
89			port_list_name: str = None,
90			port_list_id: str = None,
91			debug: bool = False,
92			):
93			targets = gmp.get_targets()
94			counter = 0
95			exists = True
96			# iterate over existing targets and find a vacant targetName
97			while exists:
98			exists = False
99			if target_name is None:
100			target_name = "target_{}".format(str(counter))
101			for target in targets.xpath('target'):
102			name = target.xpath('name/text()')[0]
103			if name == target_name:
104			exists = True
105			break
106			counter += 1
107
108			if debug:
109			print("target name: {}".format(target_name))
110
111			if not port_list_id:
112			# iterate over existing port lists and find a vacant name
113			existing_port_lists = [""]
114			for name in gmp.get_port_lists().findall("port_list/name"):
115			existing_port_lists.append(str(name.text))
116
117			if port_list_name is None:
118			port_list_name = "portlist_{}".format(str(counter))
119			counter = 0
120			while True:
121			if port_list_name not in existing_port_lists:
122			break
123			counter += 1
124
125			# create port list
126			portlist = gmp.create_port_list(portlist_name, ports)
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-12-07 08:53 UTC by Report Bug Copy Issue Report The variable `portlist_name` does not seem to be defined. Loading history...
127			port_list_id = portlist.xpath('@id')[0]
128			if debug:
129			print("New Portlist-name:\t{}".format(str(portlist_name)))
130			print("New Portlist-id:\t{}".format(str(portlist_id)))
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-12-07 08:53 UTC by Report Bug Copy Issue Report The variable `portlist_id` does not seem to be defined. Loading history...
131
132			# integrate port list id into create_target
133			res = gmp.create_target(target_name, hosts=hosts, port_list_id=port_list_id)
134			return res.xpath('@id')[0]
135
136
137			def get_alert(
138			gmp,
139			sender_email: str,
140			recipient_email: str,
141			alert_name: str = None,
142			debug=False,
143			):
144
145			# create alert if necessary
146			alert_object = gmp.get_alerts(filter='name={}'.format(alert_name))
147			alert = alert_object.xpath('alert')
148
149			if len(alert) == 0:
150			print("creating new alert {}".format(alert_name))
151			gmp.create_alert(
152			alert_name,
153			event=gmp.types.AlertEvent.TASK_RUN_STATUS_CHANGED,
154			event_data={"status": "Done"},
155			condition=gmp.types.AlertCondition.ALWAYS,
156			method=gmp.types.AlertMethod.EMAIL,
157			method_data={
158			"""Task '$n': $e
159
160			After the event $e,
161			the following condition was met: $c
162
163			This email escalation is configured to attach report format '$r'.
164			Full details and other report formats are available on the scan engine.
165
166			$t
167
168			Note:
169			This email was sent to you as a configured security scan escalation.
170			Please contact your local system administrator if you think you
171			should not have received it.
172			""": "message",
173			"2": "notice",
174			sender_email: "from_address",
175			"[OpenVAS-Manager] Task": "subject",
176			"c402cc3e-b531-11e1-9163-406186ea4fc5": "notice_attach_format",
177			recipient_email: "to_address",
178			},
179			)
180
181			alert_object = gmp.get_alerts(filter='name={}'.format(recipient_email))
182			alert = alert_object.xpath('alert')
183
184			alert_id = alert[0].get('id', 'no id found')
185			if debug:
186			print("alert_id: {}".format(str(alert_id)))
187
188			return alert_id
189
190
191			def get_scanner(gmp):
192			res = gmp.get_scanners()
193			scanner_ids = res.xpath('scanner/@id')
194			return scanner_ids[1] # default scanner
195
196
197			def create_and_start_task(
198			gmp,
199			config_id: str,
200			target_id: str,
201			scanner_id: str,
202			alert_id: str,
203			alert_name: str,
204			debug: bool = False,
205			):
206			# Create the task
207			tasks = gmp.get_tasks(filter="name~{}".format(alert_name))
208			task_name = "Alert Scan for Alert {}".format(alert_name)
209			if tasks:
210			task_name = "Alert Scan for Alert {} ({})".format(
211			alert_name, len(tasks.xpath('tasks/@id'))
212			)
213			task_comment = "Alert Scan"
214			res = gmp.create_task(
215			task_name,
216			config_id,
217			target_id,
218			scanner_id,
219			alert_ids=[alert_id],
220			comment=task_comment,
221			)
222
223			# Start the task
224			task_id = res.xpath('@id')[0]
225			gmp.start_task(task_id)
226
227			print('Task started: ' + task_name)
228
229			if debug:
230			# Stop the task (for performance reasons)
231			gmp.stop_task(task_id)
232			print('Task stopped')
233
234
235			def main(gmp, args):
236			# pylint: disable=undefined-variable
237
238			parser = ArgumentParser(
239			prefix_chars="+",
240			add_help=False,
241			formatter_class=RawTextHelpFormatter,
242			description=HELP_TEXT,
243			)
244
245			parser.add_argument(
246			"+h",
247			"++help",
248			action="help",
249			help="Show this help message and exit.",
250			)
251
252			target = parser.add_mutually_exclusive_group(required=True)
253
254			target.add_argument(
255			"++target-id",
256			type=str,
257			dest="target_id",
258			help="Use an existing by target id",
259			)
260
261			target.add_argument(
262			"++target-name",
263			type=str,
264			dest="target_name",
265			help="Create a target by name",
266			)
267
268			parser.add_argument(
269			"++hosts",
270			nargs='+',
271			dest='hosts',
272			help="Host(s) for the new target",
273			)
274
275			ports = parser.add_mutually_exclusive_group()
276
277			ports.add_argument(
278			"++port-list-id",
279			type=str,
280			dest="port_list_id",
281			help="An existing portlist id for the new target",
282			)
283			ports.add_argument(
284			"++ports",
285			type=str,
286			dest='ports',
287			help="Ports in the new target: e.g. T:80-80,8080",
288			)
289
290			parser.add_argument(
291			"++port-list-name",
292			type=str,
293			dest="port_list_name",
294			help="Name for the new portlist in the new target",
295			)
296
297			parser.add_argument(
298			"+C",
299			"++scan_config",
300			default=0,
301			type=int,
302			dest='config',
303			help="Choose from existing scan config:"
304			" 0: Full and fast"
305			" 1: Full and fast ultimate"
306			" 2: Full and very deep"
307			" 3: Full and very deep ultimate"
308			" 4: System Discovery",
309			)
310
311			parser.add_argument(
312			"++recipient",
313			required=True,
314			dest='recipient_email',
315			type=str,
316			help="Alert recipient E-Mail address",
317			)
318
319			parser.add_argument(
320			"++sender",
321			required=True,
322			dest='sender_email',
323			type=str,
324			help="Alert senders E-Mail address",
325			)
326
327			parser.add_argument(
328			"++alert-name",
329			dest='alert_name',
330			type=str,
331			help="Optional Alert name",
332			)
333
334			script_args, _ = parser.parse_known_args()
335
336			if script_args.alert_name is None:
337			script_args.alert_name = script_args.recipient_email
338
339			config_id = get_config(gmp, script_args.config)
340			if not script_args.target_id:
341			target_id = get_target(
342			gmp,
343			target_name=script_args.target_id,
344			hosts=script_args.hosts,
345			ports=script_args.ports,
346			port_list_name=script_args.port_list_name,
347			port_list_id=script_args.port_list_id,
348			)
349			else:
350			target_id = script_args.target_id
351			alert_id = get_alert(
352			gmp,
353			script_args.sender_email,
354			script_args.recipient_email,
355			script_args.alert_name,
356			)
357			scanner_id = get_scanner(gmp)
358
359			create_and_start_task(
360			gmp, config_id, target_id, scanner_id, alert_id, script_args.alert_name
361			)
362
363			print("\nScript finished\n")
364
365
366			if __name__ == '__gmp__':
367			main(gmp, args)
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-03-19 10:34 UTC by Report Bug Copy Issue Report The variable `args` does not seem to be defined. Loading history... Comprehensibility Best Practice introduced 2020-03-19 10:34 UTC by Report Bug Copy Issue Report The variable `gmp` does not seem to be defined. Loading history...
368

greenbone / gvm-tools

Pull Request — master (#344)

start-alert-scan.gmp.main() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like