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-07 08:52 UTC

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

↳ Parent: start-alert-scan.gmp

Complexity

Conditions

Size

Total Lines	128
Code Lines	89

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	89
nop	2
dl	0
loc	128
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/>.

from typing import List
from argparse import ArgumentParser, RawTextHelpFormatter


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

        usage: 
            ... 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
        
        Example:
            $ gvm-script --gmp-username name --gmp-password pass \
ssh --hostname <gsm> scripts/start-alert-scan.gmp.py <sender_email> <receiver_email>
    """


def get_config(gmp, config, debug=False):

    # get all configs of the openvas instance
    res = gmp.get_configs()

    # 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, target_id, scanner_id, alert_id, debug=False
):
    # Create the task
    tasks = gmp.get_tasks(filter="name~ScanDoneAlert")
    task_name = "ScanDoneAlert{0}".format(len(tasks.xpath('tasks/@id')))
    task_comment = "Scan Done Alert"
    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

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

    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()

    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
    if script_args.alert_name is None:
        script_args.alert_name = script_args.recipient_email
    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)

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

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