create-monthly-report.gmp.main() - Code Metrics - Inspection of "Improving monthly report and combine report script..." - greenbone/gvm-tools - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#370)

by Jaspar

created 2021-01-19 14:26 UTC

create-monthly-report.gmp.main() A

↳ Parent: create-monthly-report.gmp

Complexity

Conditions

Size

Total Lines	35
Code Lines	20

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	20
nop	2
dl	0
loc	35
rs	9.4
c	0
b	0
f	0

# -*- coding: utf-8 -*-
# Copyright (C) 2021 Greenbone Networks GmbH
#
# 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 uuid import UUID
from typing import List
from datetime import date, timedelta
from argparse import ArgumentParser, RawTextHelpFormatter

from gvmtools.helper import generate_uuid, error_and_exit

from lxml import etree as e

from gvm.xml import pretty_print

HELP_TEXT = ''


def get_last_reports_from_tasks(gmp, from_date, to_date, tags: List):
    """ Get the last reports from the tasks in the given time period """

    task_filter = 'rows=-1 '
    period_filter = 'created>{0} and created<{1}'.format(
        from_date.isoformat(), to_date.isoformat()
    )
    filter_parts = []
    if tags:
        for tag in tags:
            filter_parts.append('{} and {}'.format(period_filter, tag))

        tags_filter = ' or '.join(filter_parts)
        task_filter += tags_filter
    else:
        task_filter += period_filter

    print('Filtering the task with the filter term [{}]'.format(task_filter))

    tasks_xml = gmp.get_tasks(filter=task_filter)
    reports = []
    for report in tasks_xml.xpath('task/last_report/report/@id'):
        reports.append(str(report))

    print(reports)
    return reports


def create_filter(gmp, filter_term, date):
    filter_name = 'Filter for Monthly Report ({})'.format(date)

    res = gmp.create_filter(term=filter_term, name=filter_name)
    return res.xpath('//@id')[0]


def combine_reports(gmp, reports: List, filter_term: str):
    """ Combining the filtered ports, results and hosts of the given report ids into one new report."""
    new_uuid = generate_uuid()
    combined_report = e.Element(
        'report',
        {
            'id': new_uuid,
            'format_id': 'd5da9f67-8551-4e51-807b-b6a873d70e34',
            'extension': 'xml',
            'content_type': 'text/xml',
        },
    )
    report_elem = e.Element('report', {'id': new_uuid})

    ports_elem = e.Element('ports', {'start': '1', 'max': '-1'})
    results_elem = e.Element('results', {'start': '1', 'max': '-1'})
    combined_report.append(report_elem)
    report_elem.append(results_elem)

    hosts = []
    for report in reports:
        current_report = gmp.get_report(
            report, filter=filter_term, details=True
        )[0]
        pretty_print(current_report.find('report').find('result_count'))
        for port in current_report.xpath('report/ports/port'):
            ports_elem.append(port)
        for result in current_report.xpath('report/results/result'):
            results_elem.append(result)
        for host in current_report.xpath('report/host'):
            report_elem.append(host)

    return combined_report


def send_report(gmp, combined_report, period_start, period_end):
    """ Creating a container task and sending the combined report to the GSM """

    task_name = 'Consolidated Report [{} - {}]'.format(period_start, period_end)

    res = gmp.create_container_task(
        name=task_name, comment='Created with gvm-tools.'
    )

    task_id = res.xpath('//@id')[0]

    combined_report = e.tostring(combined_report)

    res = gmp.import_report(combined_report, task_id=task_id)

    return res.xpath('//@id')[0]


def parse_tags(tags: List):
    filter_tags = []
    for tag in tags:
        try:
            UUID(tag, version=4)
            filter_tags.append('tag_id="{}"'.format(tag))
        except ValueError:
            filter_tags.append('tag="{}"'.format(tag))

    return filter_tags


def parse_period(period: List):
    """ Parsing and validating the given time period """
    try:
        s_year, s_month, s_day = map(int, period[0].split('/'))
    except ValueError as e:
        error_and_exit(
            'Start date [{}] is not a correct date format:\n{}'.format(
                period[0], e.args[0]
            )
        )
    try:
        e_year, e_month, e_day = map(int, period[1].split('/'))
    except ValueError as e:
        error_and_exit(
            'End date [{}] is not a correct date format:\n{}'.format(
                period[1], e.args[0]
            )
        )

    try:
        period_start = date(s_year, s_month, s_day)
    except ValueError as e:
        error_and_exit('Start date: {}'.format(e.args[0]))

    try:
        period_end = date(e_year, e_month, e_day)
    except ValueError as e:
        error_and_exit('End date: {}'.format(e.args[0]))

    if period_end < period_start:
        error_and_exit('The start date seems to after the end date.')

    return period_start, period_end


def parse_args(args):  # pylint: disable=unused-argument
    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(
        '+p',
        '++period',
        nargs=2,
        type=str,
        required=True,
        dest='period',
        help='Choose a time period that is filtering the tasks. Use the date format YYYY/MM/DD.',
    )

    parser.add_argument(
        '+t',
        '++tags',
        nargs='+',
        type=str,
        dest='tags',
        help=(
            'Filter the tasks by given tag(s).\n'
            'If you pass more than on tag, they will be concatenated with '
            or '\n'
            'You can pass tag names, tag ids or tag name=value to this argument'
        ),
    )

    parser.add_argument(
        '+f',
        '++filter',
        nargs='+',
        type=str,
        dest='filter',
        help='Filter the results by given filter(s).',
    )

    script_args, _ = parser.parse_known_args()
    return script_args


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

    parsed_args = parse_args(args)

    period_start, period_end = parse_period(parsed_args.period)

    print(
        'Combining reports from tasks within the time period [{}, {}]'.format(
            period_start, period_end
        )
    )

    filter_term = ''
    if parsed_args.filter:
        filter_term = ' '.join(parsed_args.filter)
        print(
            'Filtering the results by the following filter term [{}]'.format(
                filter_term
            )
        )
    else:
        print('No result filter given.')

    filter_tags = None
    if parsed_args.tags:
        filter_tags = parse_tags(parsed_args.tags)

    reports = get_last_reports_from_tasks(
        gmp, period_start, period_end, filter_tags
    )

    combined_report = combine_reports(gmp, reports, filter_term)

    send_report(gmp, combined_report, period_start, period_end)


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



1			# -- coding: utf-8 --
2			# Copyright (C) 2021 Greenbone Networks GmbH
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 uuid import UUID
20			from typing import List
21			from datetime import date, timedelta
22			from argparse import ArgumentParser, RawTextHelpFormatter
23
24			from gvmtools.helper import generate_uuid, error_and_exit
25
26			from lxml import etree as e
27
28			from gvm.xml import pretty_print
29
30			HELP_TEXT = ''
31
32
33			def get_last_reports_from_tasks(gmp, from_date, to_date, tags: List):
34			""" Get the last reports from the tasks in the given time period """
35
36			task_filter = 'rows=-1 '
37			period_filter = 'created>{0} and created<{1}'.format(
38			from_date.isoformat(), to_date.isoformat()
39			)
40			filter_parts = []
41			if tags:
42			for tag in tags:
43			filter_parts.append('{} and {}'.format(period_filter, tag))
44
45			tags_filter = ' or '.join(filter_parts)
46			task_filter += tags_filter
47			else:
48			task_filter += period_filter
49
50			print('Filtering the task with the filter term [{}]'.format(task_filter))
51
52			tasks_xml = gmp.get_tasks(filter=task_filter)
53			reports = []
54			for report in tasks_xml.xpath('task/last_report/report/@id'):
55			reports.append(str(report))
56
57			print(reports)
58			return reports
59
60
61			def create_filter(gmp, filter_term, date):
62			filter_name = 'Filter for Monthly Report ({})'.format(date)
63
64			res = gmp.create_filter(term=filter_term, name=filter_name)
65			return res.xpath('//@id')[0]
66
67
68			def combine_reports(gmp, reports: List, filter_term: str):
69			""" Combining the filtered ports, results and hosts of the given report ids into one new report."""
70			new_uuid = generate_uuid()
71			combined_report = e.Element(
72			'report',
73			{
74			'id': new_uuid,
75			'format_id': 'd5da9f67-8551-4e51-807b-b6a873d70e34',
76			'extension': 'xml',
77			'content_type': 'text/xml',
78			},
79			)
80			report_elem = e.Element('report', {'id': new_uuid})
81
82			ports_elem = e.Element('ports', {'start': '1', 'max': '-1'})
83			results_elem = e.Element('results', {'start': '1', 'max': '-1'})
84			combined_report.append(report_elem)
85			report_elem.append(results_elem)
86
87			hosts = []
88			for report in reports:
89			current_report = gmp.get_report(
90			report, filter=filter_term, details=True
91			)[0]
92			pretty_print(current_report.find('report').find('result_count'))
93			for port in current_report.xpath('report/ports/port'):
94			ports_elem.append(port)
95			for result in current_report.xpath('report/results/result'):
96			results_elem.append(result)
97			for host in current_report.xpath('report/host'):
98			report_elem.append(host)
99
100			return combined_report
101
102
103			def send_report(gmp, combined_report, period_start, period_end):
104			""" Creating a container task and sending the combined report to the GSM """
105
106			task_name = 'Consolidated Report [{} - {}]'.format(period_start, period_end)
107
108			res = gmp.create_container_task(
109			name=task_name, comment='Created with gvm-tools.'
110			)
111
112			task_id = res.xpath('//@id')[0]
113
114			combined_report = e.tostring(combined_report)
115
116			res = gmp.import_report(combined_report, task_id=task_id)
117
118			return res.xpath('//@id')[0]
119
120
121			def parse_tags(tags: List):
122			filter_tags = []
123			for tag in tags:
124			try:
125			UUID(tag, version=4)
126			filter_tags.append('tag_id="{}"'.format(tag))
127			except ValueError:
128			filter_tags.append('tag="{}"'.format(tag))
129
130			return filter_tags
131
132
133			def parse_period(period: List):
134			""" Parsing and validating the given time period """
135			try:
136			s_year, s_month, s_day = map(int, period[0].split('/'))
137			except ValueError as e:
138			error_and_exit(
139			'Start date [{}] is not a correct date format:\n{}'.format(
140			period[0], e.args[0]
141			)
142			)
143			try:
144			e_year, e_month, e_day = map(int, period[1].split('/'))
145			except ValueError as e:
146			error_and_exit(
147			'End date [{}] is not a correct date format:\n{}'.format(
148			period[1], e.args[0]
149			)
150			)
151
152			try:
153			period_start = date(s_year, s_month, s_day)
154			except ValueError as e:
155			error_and_exit('Start date: {}'.format(e.args[0]))
156
157			try:
158			period_end = date(e_year, e_month, e_day)
159			except ValueError as e:
160			error_and_exit('End date: {}'.format(e.args[0]))
161
162			if period_end < period_start:
163			error_and_exit('The start date seems to after the end date.')
164
165			return period_start, period_end
166
167
168			def parse_args(args): # pylint: disable=unused-argument
169			parser = ArgumentParser(
170			prefix_chars='+',
171			add_help=False,
172			formatter_class=RawTextHelpFormatter,
173			description=HELP_TEXT,
174			)
175
176			parser.add_argument(
177			'+h',
178			'++help',
179			action='help',
180			help='Show this help message and exit.',
181			)
182
183			parser.add_argument(
184			'+p',
185			'++period',
186			nargs=2,
187			type=str,
188			required=True,
189			dest='period',
190			help='Choose a time period that is filtering the tasks. Use the date format YYYY/MM/DD.',
191			)
192
193			parser.add_argument(
194			'+t',
195			'++tags',
196			nargs='+',
197			type=str,
198			dest='tags',
199			help=(
200			'Filter the tasks by given tag(s).\n'
201			'If you pass more than on tag, they will be concatenated with '
202			or '\n'
203			'You can pass tag names, tag ids or tag name=value to this argument'
204			),
205			)
206
207			parser.add_argument(
208			'+f',
209			'++filter',
210			nargs='+',
211			type=str,
212			dest='filter',
213			help='Filter the results by given filter(s).',
214			)
215
216			script_args, _ = parser.parse_known_args()
217			return script_args
218
219
220			def main(gmp, args):
221			# pylint: disable=undefined-variable
222
223			parsed_args = parse_args(args)
224
225			period_start, period_end = parse_period(parsed_args.period)
226
227			print(
228			'Combining reports from tasks within the time period [{}, {}]'.format(
229			period_start, period_end
230			)
231			)
232
233			filter_term = ''
234			if parsed_args.filter:
235			filter_term = ' '.join(parsed_args.filter)
236			print(
237			'Filtering the results by the following filter term [{}]'.format(
238			filter_term
239			)
240			)
241			else:
242			print('No result filter given.')
243
244			filter_tags = None
245			if parsed_args.tags:
246			filter_tags = parse_tags(parsed_args.tags)
247
248			reports = get_last_reports_from_tasks(
249			gmp, period_start, period_end, filter_tags
250			)
251
252			combined_report = combine_reports(gmp, reports, filter_term)
253
254			send_report(gmp, combined_report, period_start, period_end)
255
256
257			if __name__ == '__gmp__':
258			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...
259

greenbone / gvm-tools

Pull Request — master (#370)

create-monthly-report.gmp.main() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like