ospd.misc.main() - Code Metrics - Inspection of "Refactor misc" - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#111)

unknown

created 2019-06-24 07:16 UTC

ospd.misc.main() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	52
Code Lines	36

Duplication

Lines	43
Ratio	82.69 %

Importance

Changes

Metric	Value
cc	5
eloc	36
nop	2
dl	43
loc	52
rs	8.5493
c	0
b	0
f	0

How to fix Long Method

# Copyright (C) 2014-2018 Greenbone Networks GmbH
#
# SPDX-License-Identifier: GPL-2.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 2
# 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, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

# pylint: disable=too-many-lines

""" Miscellaneous classes and functions related to OSPD.
"""

import logging
import os
import sys
import time
import uuid
import multiprocessing

from enum import Enum

from ospd.network import target_str_to_list

LOGGER = logging.getLogger(__name__)


class ScanStatus(Enum):
    """Scan status. """

    INIT = 0
    RUNNING = 1
    STOPPED = 2
    FINISHED = 3


class ScanCollection(object):

    """ Scans collection, managing scans and results read and write, exposing
    only needed information.

    Each scan has meta-information such as scan ID, current progress (from 0 to
    100), start time, end time, scan target and options and a list of results.

    There are 4 types of results: Alarms, Logs, Errors and Host Details.

    Todo:
    - Better checking for Scan ID existence and handling otherwise.
    - More data validation.
    - Mutex access per table/scan_info.

    """

    def __init__(self):
        """ Initialize the Scan Collection. """

        self.data_manager = None
        self.scans_table = dict()

    def add_result(
        self,
        scan_id,
        result_type,
        host='',
        hostname='',
        name='',
        value='',
        port='',
        test_id='',
        severity='',
        qod='',
    ):
        """ Add a result to a scan in the table. """

        assert scan_id
        assert len(name) or len(value)
        result = dict()
        result['type'] = result_type
        result['name'] = name
        result['severity'] = severity
        result['test_id'] = test_id
        result['value'] = value
        result['host'] = host
        result['hostname'] = hostname
        result['port'] = port
        result['qod'] = qod
        results = self.scans_table[scan_id]['results']
        results.append(result)
        # Set scan_info's results to propagate results to parent process.
        self.scans_table[scan_id]['results'] = results

    def set_progress(self, scan_id, progress):
        """ Sets scan_id scan's progress. """

        if progress > 0 and progress <= 100:
            self.scans_table[scan_id]['progress'] = progress
        if progress == 100:
            self.scans_table[scan_id]['end_time'] = int(time.time())

    def set_target_progress(self, scan_id, target, host, progress):
        """ Sets scan_id scan's progress. """
        if progress > 0 and progress <= 100:
            targets = self.scans_table[scan_id]['target_progress']
            targets[target][host] = progress
            # Set scan_info's target_progress to propagate progresses
            # to parent process.
            self.scans_table[scan_id]['target_progress'] = targets

    def set_host_finished(self, scan_id, target, host):
        """ Add the host in a list of finished hosts """
        finished_hosts = self.scans_table[scan_id]['finished_hosts']
        finished_hosts[target].append(host)
        self.scans_table[scan_id]['finished_hosts'] = finished_hosts

    def get_hosts_unfinished(self, scan_id):
        """ Get a list of unfinished hosts."""

        unfinished_hosts = list()
        for target in self.scans_table[scan_id]['finished_hosts']:
            unfinished_hosts.extend(target_str_to_list(target))
        for target in self.scans_table[scan_id]['finished_hosts']:
            for host in self.scans_table[scan_id]['finished_hosts'][target]:
                unfinished_hosts.remove(host)

        return unfinished_hosts

    def get_hosts_finished(self, scan_id):
        """ Get a list of finished hosts."""

        finished_hosts = list()
        for target in self.scans_table[scan_id]['finished_hosts']:
            finished_hosts.extend(
                self.scans_table[scan_id]['finished_hosts'].get(target)
            )

        return finished_hosts

    def results_iterator(self, scan_id, pop_res):
        """ Returns an iterator over scan_id scan's results. If pop_res is True,
        it removed the fetched results from the list.
        """
        if pop_res:
            result_aux = self.scans_table[scan_id]['results']
            self.scans_table[scan_id]['results'] = list()
            return iter(result_aux)

        return iter(self.scans_table[scan_id]['results'])

    def ids_iterator(self):
        """ Returns an iterator over the collection's scan IDS. """

        return iter(self.scans_table.keys())

    def remove_single_result(self, scan_id, result):
        """Removes a single result from the result list in scan_table.

        Parameters:
            scan_id (uuid): Scan ID to identify the scan process to be resumed.
            result (dict): The result to be removed from the results list.
        """
        results = self.scans_table[scan_id]['results']
        results.remove(result)
        self.scans_table[scan_id]['results'] = results

    def del_results_for_stopped_hosts(self, scan_id):
        """ Remove results from the result table for those host
        """
        unfinished_hosts = self.get_hosts_unfinished(scan_id)
        for result in self.results_iterator(scan_id, False):
            if result['host'] in unfinished_hosts:
                self.remove_single_result(scan_id, result)

    def resume_scan(self, scan_id, options):
        """ Reset the scan status in the scan_table to INIT.
        Also, overwrite the options, because a resume task cmd
        can add some new option. E.g. exclude hosts list.
        Parameters:
            scan_id (uuid): Scan ID to identify the scan process to be resumed.
            options (dict): Options for the scan to be resumed. This options
                            are not added to the already existent ones.
                            The old ones are removed

        Return:
            Scan ID which identifies the current scan.
        """
        self.scans_table[scan_id]['status'] = ScanStatus.INIT
        if options:
            self.scans_table[scan_id]['options'] = options

        self.del_results_for_stopped_hosts(scan_id)

        return scan_id

    def create_scan(self, scan_id='', targets='', options=None, vts=''):
        """ Creates a new scan with provided scan information. """

        if self.data_manager is None:
            self.data_manager = multiprocessing.Manager()

        # Check if it is possible to resume task. To avoid to resume, the
        # scan must be deleted from the scans_table.
        if (
            scan_id
            and self.id_exists(scan_id)
            and (self.get_status(scan_id) == ScanStatus.STOPPED)
        ):
            return self.resume_scan(scan_id, options)

        if not options:
            options = dict()
        scan_info = self.data_manager.dict()
        scan_info['results'] = list()
        scan_info['finished_hosts'] = dict(
            [[target, []] for target, _, _, _ in targets]
        )
        scan_info['progress'] = 0
        scan_info['target_progress'] = dict(
            [[target, {}] for target, _, _, _ in targets]
        )
        scan_info['targets'] = targets
        scan_info['vts'] = vts
        scan_info['options'] = options
        scan_info['start_time'] = int(time.time())
        scan_info['end_time'] = "0"
        scan_info['status'] = ScanStatus.INIT
        if scan_id is None or scan_id == '':
            scan_id = str(uuid.uuid4())
        scan_info['scan_id'] = scan_id
        self.scans_table[scan_id] = scan_info
        return scan_id

    def set_status(self, scan_id, status):
        """ Sets scan_id scan's status. """
        self.scans_table[scan_id]['status'] = status

    def get_status(self, scan_id):
        """ Get scan_id scans's status."""

        return self.scans_table[scan_id]['status']

    def get_options(self, scan_id):
        """ Get scan_id scan's options list. """

        return self.scans_table[scan_id]['options']

    def set_option(self, scan_id, name, value):
        """ Set a scan_id scan's name option to value. """

        self.scans_table[scan_id]['options'][name] = value

    def get_progress(self, scan_id):
        """ Get a scan's current progress value. """

        return self.scans_table[scan_id]['progress']

    def get_target_progress(self, scan_id, target):
        """ Get a target's current progress value.
        The value is calculated with the progress of each single host
        in the target."""

        total_hosts = len(target_str_to_list(target))
        host_progresses = self.scans_table[scan_id]['target_progress'].get(
            target
        )
        try:
            t_prog = sum(host_progresses.values()) / total_hosts
        except ZeroDivisionError:
            LOGGER.error(
                "Zero division error in %s", self.get_target_progress.__name__
            )
            raise
        return t_prog

    def get_start_time(self, scan_id):
        """ Get a scan's start time. """

        return self.scans_table[scan_id]['start_time']

    def get_end_time(self, scan_id):
        """ Get a scan's end time. """

        return self.scans_table[scan_id]['end_time']

    def get_target_list(self, scan_id):
        """ Get a scan's target list. """

        target_list = []
        for target, _, _, _ in self.scans_table[scan_id]['targets']:
            target_list.append(target)
        return target_list

    def get_ports(self, scan_id, target):
        """ Get a scan's ports list. If a target is specified
        it will return the corresponding port for it. If not,
        it returns the port item of the first nested list in
        the target's list.
        """
        if target:
            for item in self.scans_table[scan_id]['targets']:
                if target == item[0]:
                    return item[1]

        return self.scans_table[scan_id]['targets'][0][1]

    def get_exclude_hosts(self, scan_id, target):
        """ Get an exclude host list for a given target.
        """
        if target:
            for item in self.scans_table[scan_id]['targets']:
                if target == item[0]:
                    return item[3]

    def get_credentials(self, scan_id, target):
        """ Get a scan's credential list. It return dictionary with
        the corresponding credential for a given target.
        """
        if target:
            for item in self.scans_table[scan_id]['targets']:
                if target == item[0]:
                    return item[2]

    def get_vts(self, scan_id):
        """ Get a scan's vts list. """

        return self.scans_table[scan_id]['vts']

    def id_exists(self, scan_id):
        """ Check whether a scan exists in the table. """

        return self.scans_table.get(scan_id) is not None

    def delete_scan(self, scan_id):
        """ Delete a scan if fully finished. """

        if self.get_status(scan_id) == ScanStatus.RUNNING:
            return False
        self.scans_table.pop(scan_id)
        if len(self.scans_table) == 0:
            del self.data_manager
            self.data_manager = None
        return True


class ResultType(object):

    """ Various scan results types values. """

    ALARM = 0
    LOG = 1
    ERROR = 2
    HOST_DETAIL = 3

    @classmethod
    def get_str(cls, result_type):
        """ Return string name of a result type. """
        if result_type == cls.ALARM:
            return "Alarm"
        elif result_type == cls.LOG:
            return "Log Message"
        elif result_type == cls.ERROR:
            return "Error Message"
        elif result_type == cls.HOST_DETAIL:
            return "Host Detail"
        else:
            assert False, "Erroneous result type {0}.".format(result_type)

    @classmethod
    def get_type(cls, result_name):
        """ Return string name of a result type. """
        if result_name == "Alarm":
            return cls.ALARM
        elif result_name == "Log Message":
            return cls.LOG
        elif result_name == "Error Message":
            return cls.ERROR
        elif result_name == "Host Detail":
            return cls.HOST_DETAIL
        else:
            assert False, "Erroneous result name {0}.".format(result_name)


def valid_uuid(value):
    """ Check if value is a valid UUID. """

    try:
        uuid.UUID(value, version=4)
        return True
    except (TypeError, ValueError, AttributeError):
        return False


def go_to_background():
    """ Daemonize the running process. """
    try:
        if os.fork():
            sys.exit()
    except OSError as errmsg:
        LOGGER.error('Fork failed: %s', errmsg)
        sys.exit(1)


1			# Copyright (C) 2014-2018 Greenbone Networks GmbH
2			#
3			# SPDX-License-Identifier: GPL-2.0-or-later
4			#
5			# This program is free software; you can redistribute it and/or
6			# modify it under the terms of the GNU General Public License
7			# as published by the Free Software Foundation; either version 2
8			# of the License, or (at your option) any later version.
9			#
10			# This program is distributed in the hope that it will be useful,
11			# but WITHOUT ANY WARRANTY; without even the implied warranty of
12			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13			# GNU General Public License for more details.
14			#
15			# You should have received a copy of the GNU General Public License
16			# along with this program; if not, write to the Free Software
17			# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
18
19			# pylint: disable=too-many-lines
20
21			""" Miscellaneous classes and functions related to OSPD.
22			"""
23
24			import logging
25			import os
26			import sys
27			import time
28			import uuid
29			import multiprocessing
30
31			from enum import Enum
32
33			from ospd.network import target_str_to_list
34
35			LOGGER = logging.getLogger(__name__)
36
37
38			class ScanStatus(Enum):
39			"""Scan status. """
40
41			INIT = 0
42			RUNNING = 1
43			STOPPED = 2
44			FINISHED = 3
45
46
47			class ScanCollection(object):
48
49			""" Scans collection, managing scans and results read and write, exposing
50			only needed information.
51
52			Each scan has meta-information such as scan ID, current progress (from 0 to
53			100), start time, end time, scan target and options and a list of results.
54
55			There are 4 types of results: Alarms, Logs, Errors and Host Details.
56
57			Todo:
58			- Better checking for Scan ID existence and handling otherwise.
59			- More data validation.
60			- Mutex access per table/scan_info.
61
62			"""
63
64			def __init__(self):
65			""" Initialize the Scan Collection. """
66
67			self.data_manager = None
68			self.scans_table = dict()
69
70			def add_result(
71			self,
72			scan_id,
73			result_type,
74			host='',
75			hostname='',
76			name='',
77			value='',
78			port='',
79			test_id='',
80			severity='',
81			qod='',
82			):
83			""" Add a result to a scan in the table. """
84
85			assert scan_id
86			assert len(name) or len(value)
87			result = dict()
88			result['type'] = result_type
89			result['name'] = name
90			result['severity'] = severity
91			result['test_id'] = test_id
92			result['value'] = value
93			result['host'] = host
94			result['hostname'] = hostname
95			result['port'] = port
96			result['qod'] = qod
97			results = self.scans_table[scan_id]['results']
98			results.append(result)
99			# Set scan_info's results to propagate results to parent process.
100			self.scans_table[scan_id]['results'] = results
101
102			def set_progress(self, scan_id, progress):
103			""" Sets scan_id scan's progress. """
104
105			if progress > 0 and progress <= 100:
106			self.scans_table[scan_id]['progress'] = progress
107			if progress == 100:
108			self.scans_table[scan_id]['end_time'] = int(time.time())
109
110			def set_target_progress(self, scan_id, target, host, progress):
111			""" Sets scan_id scan's progress. """
112			if progress > 0 and progress <= 100:
113			targets = self.scans_table[scan_id]['target_progress']
114			targets[target][host] = progress
115			# Set scan_info's target_progress to propagate progresses
116			# to parent process.
117			self.scans_table[scan_id]['target_progress'] = targets
118
119			def set_host_finished(self, scan_id, target, host):
120			""" Add the host in a list of finished hosts """
121			finished_hosts = self.scans_table[scan_id]['finished_hosts']
122			finished_hosts[target].append(host)
123			self.scans_table[scan_id]['finished_hosts'] = finished_hosts
124
125			def get_hosts_unfinished(self, scan_id):
126			""" Get a list of unfinished hosts."""
127
128			unfinished_hosts = list()
129			for target in self.scans_table[scan_id]['finished_hosts']:
130			unfinished_hosts.extend(target_str_to_list(target))
131			for target in self.scans_table[scan_id]['finished_hosts']:
132			for host in self.scans_table[scan_id]['finished_hosts'][target]:
133			unfinished_hosts.remove(host)
134
135			return unfinished_hosts
136
137			def get_hosts_finished(self, scan_id):
138			""" Get a list of finished hosts."""
139
140			finished_hosts = list()
141			for target in self.scans_table[scan_id]['finished_hosts']:
142			finished_hosts.extend(
143			self.scans_table[scan_id]['finished_hosts'].get(target)
144			)
145
146			return finished_hosts
147
148			def results_iterator(self, scan_id, pop_res):
149			""" Returns an iterator over scan_id scan's results. If pop_res is True,
150			it removed the fetched results from the list.
151			"""
152			if pop_res:
153			result_aux = self.scans_table[scan_id]['results']
154			self.scans_table[scan_id]['results'] = list()
155			return iter(result_aux)
156
157			return iter(self.scans_table[scan_id]['results'])
158
159			def ids_iterator(self):
160			""" Returns an iterator over the collection's scan IDS. """
161
162			return iter(self.scans_table.keys())
163
164			def remove_single_result(self, scan_id, result):
165			"""Removes a single result from the result list in scan_table.
166
167			Parameters:
168			scan_id (uuid): Scan ID to identify the scan process to be resumed.
169			result (dict): The result to be removed from the results list.
170			"""
171			results = self.scans_table[scan_id]['results']
172			results.remove(result)
173			self.scans_table[scan_id]['results'] = results
174
175			def del_results_for_stopped_hosts(self, scan_id):
176			""" Remove results from the result table for those host
177			"""
178			unfinished_hosts = self.get_hosts_unfinished(scan_id)
179			for result in self.results_iterator(scan_id, False):
180			if result['host'] in unfinished_hosts:
181			self.remove_single_result(scan_id, result)
182
183			def resume_scan(self, scan_id, options):
184			""" Reset the scan status in the scan_table to INIT.
185			Also, overwrite the options, because a resume task cmd
186			can add some new option. E.g. exclude hosts list.
187			Parameters:
188			scan_id (uuid): Scan ID to identify the scan process to be resumed.
189			options (dict): Options for the scan to be resumed. This options
190			are not added to the already existent ones.
191			The old ones are removed
192
193			Return:
194			Scan ID which identifies the current scan.
195			"""
196			self.scans_table[scan_id]['status'] = ScanStatus.INIT
197			if options:
198			self.scans_table[scan_id]['options'] = options
199
200			self.del_results_for_stopped_hosts(scan_id)
201
202			return scan_id
203
204			def create_scan(self, scan_id='', targets='', options=None, vts=''):
205			""" Creates a new scan with provided scan information. """
206
207			if self.data_manager is None:
208			self.data_manager = multiprocessing.Manager()
209
210			# Check if it is possible to resume task. To avoid to resume, the
211			# scan must be deleted from the scans_table.
212			if (
213			scan_id
214			and self.id_exists(scan_id)
215			and (self.get_status(scan_id) == ScanStatus.STOPPED)
216			):
217			return self.resume_scan(scan_id, options)
218
219			if not options:
220			options = dict()
221			scan_info = self.data_manager.dict()
222			scan_info['results'] = list()
223			scan_info['finished_hosts'] = dict(
224			[[target, []] for target, _, _, _ in targets]
225			)
226			scan_info['progress'] = 0
227			scan_info['target_progress'] = dict(
228			[[target, {}] for target, _, _, _ in targets]
229			)
230			scan_info['targets'] = targets
231			scan_info['vts'] = vts
232			scan_info['options'] = options
233			scan_info['start_time'] = int(time.time())
234			scan_info['end_time'] = "0"
235			scan_info['status'] = ScanStatus.INIT
236			if scan_id is None or scan_id == '':
237			scan_id = str(uuid.uuid4())
238			scan_info['scan_id'] = scan_id
239			self.scans_table[scan_id] = scan_info
240			return scan_id
241
242			def set_status(self, scan_id, status):
243			""" Sets scan_id scan's status. """
244			self.scans_table[scan_id]['status'] = status
245
246			def get_status(self, scan_id):
247			""" Get scan_id scans's status."""
248
249			return self.scans_table[scan_id]['status']
250
251			def get_options(self, scan_id):
252			""" Get scan_id scan's options list. """
253
254			return self.scans_table[scan_id]['options']
255
256			def set_option(self, scan_id, name, value):
257			""" Set a scan_id scan's name option to value. """
258
259			self.scans_table[scan_id]['options'][name] = value
260
261			def get_progress(self, scan_id):
262			""" Get a scan's current progress value. """
263
264			return self.scans_table[scan_id]['progress']
265
266			def get_target_progress(self, scan_id, target):
267			""" Get a target's current progress value.
268			The value is calculated with the progress of each single host
269			in the target."""
270
271			total_hosts = len(target_str_to_list(target))
272			host_progresses = self.scans_table[scan_id]['target_progress'].get(
273			target
274			)
275			try:
276			t_prog = sum(host_progresses.values()) / total_hosts
277			except ZeroDivisionError:
278			LOGGER.error(
279			"Zero division error in %s", self.get_target_progress.__name__
280			)
281			raise
282			return t_prog
283
284			def get_start_time(self, scan_id):
285			""" Get a scan's start time. """
286
287			return self.scans_table[scan_id]['start_time']
288
289			def get_end_time(self, scan_id):
290			""" Get a scan's end time. """
291
292			return self.scans_table[scan_id]['end_time']
293
294			def get_target_list(self, scan_id):
295			""" Get a scan's target list. """
296
297			target_list = []
298			for target, _, _, _ in self.scans_table[scan_id]['targets']:
299			target_list.append(target)
300			return target_list
301
302			def get_ports(self, scan_id, target):
303			""" Get a scan's ports list. If a target is specified
304			it will return the corresponding port for it. If not,
305			it returns the port item of the first nested list in
306			the target's list.
307			"""
308			if target:
309			for item in self.scans_table[scan_id]['targets']:
310			if target == item[0]:
311			return item[1]
312
313			return self.scans_table[scan_id]['targets'][0][1]
314
315			def get_exclude_hosts(self, scan_id, target):
316			""" Get an exclude host list for a given target.
317			"""
318			if target:
319			for item in self.scans_table[scan_id]['targets']:
320			if target == item[0]:
321			return item[3]
322
323			def get_credentials(self, scan_id, target):
324			""" Get a scan's credential list. It return dictionary with
325			the corresponding credential for a given target.
326			"""
327			if target:
328			for item in self.scans_table[scan_id]['targets']:
329			if target == item[0]:
330			return item[2]
331
332			def get_vts(self, scan_id):
333			""" Get a scan's vts list. """
334
335			return self.scans_table[scan_id]['vts']
336
337			def id_exists(self, scan_id):
338			""" Check whether a scan exists in the table. """
339
340			return self.scans_table.get(scan_id) is not None
341
342			def delete_scan(self, scan_id):
343			""" Delete a scan if fully finished. """
344
345			if self.get_status(scan_id) == ScanStatus.RUNNING:
346			return False
347			self.scans_table.pop(scan_id)
348			if len(self.scans_table) == 0:
349			del self.data_manager
350			self.data_manager = None
351			return True
352
353
354			class ResultType(object):
355
356			""" Various scan results types values. """
357
358			ALARM = 0
359			LOG = 1
360			ERROR = 2
361			HOST_DETAIL = 3
362
363			@classmethod
364			def get_str(cls, result_type):
365			""" Return string name of a result type. """
366			if result_type == cls.ALARM:
367			return "Alarm"
368			elif result_type == cls.LOG:
369			return "Log Message"
370			elif result_type == cls.ERROR:
371			return "Error Message"
372			elif result_type == cls.HOST_DETAIL:
373			return "Host Detail"
374			else:
375			assert False, "Erroneous result type {0}.".format(result_type)
376
377			@classmethod
378			def get_type(cls, result_name):
379			""" Return string name of a result type. """
380			if result_name == "Alarm":
381			return cls.ALARM
382			elif result_name == "Log Message":
383			return cls.LOG
384			elif result_name == "Error Message":
385			return cls.ERROR
386			elif result_name == "Host Detail":
387			return cls.HOST_DETAIL
388			else:
389			assert False, "Erroneous result name {0}.".format(result_name)
390
391
392			def valid_uuid(value):
393			""" Check if value is a valid UUID. """
394
395			try:
396			uuid.UUID(value, version=4)
397			return True
398			except (TypeError, ValueError, AttributeError):
399			return False
400
401
402			def go_to_background():
403			""" Daemonize the running process. """
404			try:
405			if os.fork():
406			sys.exit()
407			except OSError as errmsg:
408			LOGGER.error('Fork failed: %s', errmsg)
409			sys.exit(1)
410

greenbone / ospd

Pull Request — master (#111)

ospd.misc.main() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like