ospd.misc.ScanCollection.simplify_exclude_host_list() - Code Metrics - Inspection of "Remove from exclude_host list the hosts passed as..." - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#182)

by Juan José

created 2019-11-14 15:21 UTC

ScanCollection.simplify_exclude_host_list() A

↳ Parent: ospd.misc

Complexity

Conditions

Size

Total Lines	20
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	10
nop	3
dl	0
loc	20
rs	9.3333
c	0
b	0
f	0

# 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 collections import OrderedDict
from pathlib import Path

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 = OrderedDict()
        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 remove_hosts_from_target_progress(self, scan_id, target, hosts):
        """Remove a list of hosts from the main scan progress table to avoid
        the hosts to be included in the calculation of the scan progress"""
        if not hosts:
            return

        targets = self.scans_table[scan_id]['target_progress']
        for host in hosts:
            if host in targets[target]:
                del targets[target][host]

        # Set scan_info's target_progress to propagate progresses
        # to parent process.
        self.scans_table[scan_id]['target_progress'] = targets

    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_host_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']

        if host not in finished_hosts[target]:
            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)
        ):
            self.scans_table[scan_id]['end_time'] = 0

            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
        if status == ScanStatus.STOPPED:
            self.scans_table[scan_id]['end_time'] = int(time.time())

    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 simplify_exclude_host_list(self, scan_id, target):
        """ Remove from exclude_hosts the received hosts in the finished_hosts
        list sent by the client.
        The finished hosts are sent also as exclude hosts for backward
        compatibility purposses.
        """

        exc_hosts_list = target_str_to_list(
            self.get_exclude_hosts(scan_id, target)
        )

        finished_hosts_list = target_str_to_list(
            self.get_finished_hosts(scan_id, target)
        )
        if finished_hosts_list and exc_hosts_list:
            for finished in finished_hosts_list:
                if finished in exc_hosts_list:
                    exc_hosts_list.remove(finished)

        return exc_hosts_list

    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))
        exc_hosts_list = self.simplify_exclude_host_list(scan_id, target)
        exc_hosts = len(exc_hosts_list) if exc_hosts_list else 0
        host_progresses = self.scans_table[scan_id]['target_progress'].get(
            target
        )
        try:
            t_prog = sum(host_progresses.values()) / (total_hosts - exc_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_finished_hosts(self, scan_id, target):
        """ Get the finished host list sent by the client for a given target.
        """
        if target:
            for item in self.scans_table[scan_id]['targets']:
                if target == item[0]:
                    return item[4]

    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)


def create_pid(pidfile):
    """ Check if there is an already running daemon and creates the pid file.
    Otherwise gives an error. """

    pid = str(os.getpid())

    if Path(pidfile).is_file():
        LOGGER.error("There is an already running process.")
        return False

    try:
        with open(pidfile, 'w') as f:
            f.write(pid)
    except (FileNotFoundError, PermissionError) as e:
        msg = "Failed to create pid file %s. %s" % (os.path.dirname(pidfile), e)
        LOGGER.error(msg)
        return False

    return True


def remove_pidfile(pidfile, signum=None, frame=None):
    """ Removes the pidfile before ending the daemon. """
    pidpath = Path(pidfile)
    if pidpath.is_file():
        with pidpath.open() as f:
            if int(f.read()) == os.getpid():
                LOGGER.debug("Finishing daemon process")
                pidpath.unlink()
                sys.exit()


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			from collections import OrderedDict
33			from pathlib import Path
34
35			from ospd.network import target_str_to_list
36
37			LOGGER = logging.getLogger(__name__)
38
39
40			class ScanStatus(Enum):
41			"""Scan status. """
42
43			INIT = 0
44			RUNNING = 1
45			STOPPED = 2
46			FINISHED = 3
47
48
49			class ScanCollection(object):
50
51			""" Scans collection, managing scans and results read and write, exposing
52			only needed information.
53
54			Each scan has meta-information such as scan ID, current progress (from 0 to
55			100), start time, end time, scan target and options and a list of results.
56
57			There are 4 types of results: Alarms, Logs, Errors and Host Details.
58
59			Todo:
60			- Better checking for Scan ID existence and handling otherwise.
61			- More data validation.
62			- Mutex access per table/scan_info.
63
64			"""
65
66			def __init__(self):
67			""" Initialize the Scan Collection. """
68
69			self.data_manager = None
70			self.scans_table = dict()
71
72			def add_result(
73			self,
74			scan_id,
75			result_type,
76			host='',
77			hostname='',
78			name='',
79			value='',
80			port='',
81			test_id='',
82			severity='',
83			qod='',
84			):
85			""" Add a result to a scan in the table. """
86
87			assert scan_id
88			assert len(name) or len(value)
89			result = OrderedDict()
90			result['type'] = result_type
91			result['name'] = name
92			result['severity'] = severity
93			result['test_id'] = test_id
94			result['value'] = value
95			result['host'] = host
96			result['hostname'] = hostname
97			result['port'] = port
98			result['qod'] = qod
99			results = self.scans_table[scan_id]['results']
100			results.append(result)
101			# Set scan_info's results to propagate results to parent process.
102			self.scans_table[scan_id]['results'] = results
103
104			def remove_hosts_from_target_progress(self, scan_id, target, hosts):
105			"""Remove a list of hosts from the main scan progress table to avoid
106			the hosts to be included in the calculation of the scan progress"""
107			if not hosts:
108			return
109
110			targets = self.scans_table[scan_id]['target_progress']
111			for host in hosts:
112			if host in targets[target]:
113			del targets[target][host]
114
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_progress(self, scan_id, progress):
120			""" Sets scan_id scan's progress. """
121
122			if progress > 0 and progress <= 100:
123			self.scans_table[scan_id]['progress'] = progress
124			if progress == 100:
125			self.scans_table[scan_id]['end_time'] = int(time.time())
126
127			def set_host_progress(self, scan_id, target, host, progress):
128			""" Sets scan_id scan's progress. """
129			if progress > 0 and progress <= 100:
130			targets = self.scans_table[scan_id]['target_progress']
131			targets[target][host] = progress
132			# Set scan_info's target_progress to propagate progresses
133			# to parent process.
134			self.scans_table[scan_id]['target_progress'] = targets
135
136			def set_host_finished(self, scan_id, target, host):
137			""" Add the host in a list of finished hosts """
138			finished_hosts = self.scans_table[scan_id]['finished_hosts']
139
140			if host not in finished_hosts[target]:
141			finished_hosts[target].append(host)
142
143			self.scans_table[scan_id]['finished_hosts'] = finished_hosts
144
145			def get_hosts_unfinished(self, scan_id):
146			""" Get a list of unfinished hosts."""
147
148			unfinished_hosts = list()
149			for target in self.scans_table[scan_id]['finished_hosts']:
150			unfinished_hosts.extend(target_str_to_list(target))
151			for target in self.scans_table[scan_id]['finished_hosts']:
152			for host in self.scans_table[scan_id]['finished_hosts'][target]:
153			unfinished_hosts.remove(host)
154
155			return unfinished_hosts
156
157			def get_hosts_finished(self, scan_id):
158			""" Get a list of finished hosts."""
159
160			finished_hosts = list()
161			for target in self.scans_table[scan_id]['finished_hosts']:
162			finished_hosts.extend(
163			self.scans_table[scan_id]['finished_hosts'].get(target)
164			)
165
166			return finished_hosts
167
168			def results_iterator(self, scan_id, pop_res):
169			""" Returns an iterator over scan_id scan's results. If pop_res is True,
170			it removed the fetched results from the list.
171			"""
172			if pop_res:
173			result_aux = self.scans_table[scan_id]['results']
174			self.scans_table[scan_id]['results'] = list()
175			return iter(result_aux)
176
177			return iter(self.scans_table[scan_id]['results'])
178
179			def ids_iterator(self):
180			""" Returns an iterator over the collection's scan IDS. """
181
182			return iter(self.scans_table.keys())
183
184			def remove_single_result(self, scan_id, result):
185			"""Removes a single result from the result list in scan_table.
186
187			Parameters:
188			scan_id (uuid): Scan ID to identify the scan process to be resumed.
189			result (dict): The result to be removed from the results list.
190			"""
191			results = self.scans_table[scan_id]['results']
192			results.remove(result)
193			self.scans_table[scan_id]['results'] = results
194
195			def del_results_for_stopped_hosts(self, scan_id):
196			""" Remove results from the result table for those host
197			"""
198			unfinished_hosts = self.get_hosts_unfinished(scan_id)
199			for result in self.results_iterator(scan_id, False):
200			if result['host'] in unfinished_hosts:
201			self.remove_single_result(scan_id, result)
202
203			def resume_scan(self, scan_id, options):
204			""" Reset the scan status in the scan_table to INIT.
205			Also, overwrite the options, because a resume task cmd
206			can add some new option. E.g. exclude hosts list.
207			Parameters:
208			scan_id (uuid): Scan ID to identify the scan process to be resumed.
209			options (dict): Options for the scan to be resumed. This options
210			are not added to the already existent ones.
211			The old ones are removed
212
213			Return:
214			Scan ID which identifies the current scan.
215			"""
216			self.scans_table[scan_id]['status'] = ScanStatus.INIT
217			if options:
218			self.scans_table[scan_id]['options'] = options
219
220			self.del_results_for_stopped_hosts(scan_id)
221
222			return scan_id
223
224			def create_scan(self, scan_id='', targets='', options=None, vts=''):
225			""" Creates a new scan with provided scan information. """
226
227			if self.data_manager is None:
228			self.data_manager = multiprocessing.Manager()
229
230			# Check if it is possible to resume task. To avoid to resume, the
231			# scan must be deleted from the scans_table.
232			if (
233			scan_id
234			and self.id_exists(scan_id)
235			and (self.get_status(scan_id) == ScanStatus.STOPPED)
236			):
237			self.scans_table[scan_id]['end_time'] = 0
238
239			return self.resume_scan(scan_id, options)
240
241			if not options:
242			options = dict()
243			scan_info = self.data_manager.dict()
244			scan_info['results'] = list()
245			scan_info['finished_hosts'] = dict(
246			[[target, []] for target, _, _, _, _ in targets]
247			)
248			scan_info['progress'] = 0
249			scan_info['target_progress'] = dict(
250			[[target, {}] for target, _, _, _, _ in targets]
251			)
252			scan_info['targets'] = targets
253			scan_info['vts'] = vts
254			scan_info['options'] = options
255			scan_info['start_time'] = int(time.time())
256			scan_info['end_time'] = 0
257			scan_info['status'] = ScanStatus.INIT
258			if scan_id is None or scan_id == '':
259			scan_id = str(uuid.uuid4())
260			scan_info['scan_id'] = scan_id
261			self.scans_table[scan_id] = scan_info
262			return scan_id
263
264			def set_status(self, scan_id, status):
265			""" Sets scan_id scan's status. """
266			self.scans_table[scan_id]['status'] = status
267			if status == ScanStatus.STOPPED:
268			self.scans_table[scan_id]['end_time'] = int(time.time())
269
270			def get_status(self, scan_id):
271			""" Get scan_id scans's status."""
272
273			return self.scans_table[scan_id]['status']
274
275			def get_options(self, scan_id):
276			""" Get scan_id scan's options list. """
277
278			return self.scans_table[scan_id]['options']
279
280			def set_option(self, scan_id, name, value):
281			""" Set a scan_id scan's name option to value. """
282
283			self.scans_table[scan_id]['options'][name] = value
284
285			def get_progress(self, scan_id):
286			""" Get a scan's current progress value. """
287
288			return self.scans_table[scan_id]['progress']
289
290			def simplify_exclude_host_list(self, scan_id, target):
291			""" Remove from exclude_hosts the received hosts in the finished_hosts
292			list sent by the client.
293			The finished hosts are sent also as exclude hosts for backward
294			compatibility purposses.
295			"""
296
297			exc_hosts_list = target_str_to_list(
298			self.get_exclude_hosts(scan_id, target)
299			)
300
301			finished_hosts_list = target_str_to_list(
302			self.get_finished_hosts(scan_id, target)
303			)
304			if finished_hosts_list and exc_hosts_list:
305			for finished in finished_hosts_list:
306			if finished in exc_hosts_list:
307			exc_hosts_list.remove(finished)
308
309			return exc_hosts_list
310
311			def get_target_progress(self, scan_id, target):
312			""" Get a target's current progress value.
313			The value is calculated with the progress of each single host
314			in the target."""
315
316			total_hosts = len(target_str_to_list(target))
317			exc_hosts_list = self.simplify_exclude_host_list(scan_id, target)
318			exc_hosts = len(exc_hosts_list) if exc_hosts_list else 0
319			host_progresses = self.scans_table[scan_id]['target_progress'].get(
320			target
321			)
322			try:
323			t_prog = sum(host_progresses.values()) / (total_hosts - exc_hosts)
324			except ZeroDivisionError:
325			LOGGER.error(
326			"Zero division error in %s", self.get_target_progress.__name__
327			)
328			raise
329			return t_prog
330
331			def get_start_time(self, scan_id):
332			""" Get a scan's start time. """
333
334			return self.scans_table[scan_id]['start_time']
335
336			def get_end_time(self, scan_id):
337			""" Get a scan's end time. """
338
339			return self.scans_table[scan_id]['end_time']
340
341			def get_target_list(self, scan_id):
342			""" Get a scan's target list. """
343
344			target_list = []
345			for target, _, _, _, _ in self.scans_table[scan_id]['targets']:
346			target_list.append(target)
347			return target_list
348
349			def get_ports(self, scan_id, target):
350			""" Get a scan's ports list. If a target is specified
351			it will return the corresponding port for it. If not,
352			it returns the port item of the first nested list in
353			the target's list.
354			"""
355			if target:
356			for item in self.scans_table[scan_id]['targets']:
357			if target == item[0]:
358			return item[1]
359
360			return self.scans_table[scan_id]['targets'][0][1]
361
362			def get_exclude_hosts(self, scan_id, target):
363			""" Get an exclude host list for a given target.
364			"""
365			if target:
366			for item in self.scans_table[scan_id]['targets']:
367			if target == item[0]:
368			return item[3]
369
370			def get_finished_hosts(self, scan_id, target):
371			""" Get the finished host list sent by the client for a given target.
372			"""
373			if target:
374			for item in self.scans_table[scan_id]['targets']:
375			if target == item[0]:
376			return item[4]
377
378			def get_credentials(self, scan_id, target):
379			""" Get a scan's credential list. It return dictionary with
380			the corresponding credential for a given target.
381			"""
382			if target:
383			for item in self.scans_table[scan_id]['targets']:
384			if target == item[0]:
385			return item[2]
386
387			def get_vts(self, scan_id):
388			""" Get a scan's vts list. """
389
390			return self.scans_table[scan_id]['vts']
391
392			def id_exists(self, scan_id):
393			""" Check whether a scan exists in the table. """
394
395			return self.scans_table.get(scan_id) is not None
396
397			def delete_scan(self, scan_id):
398			""" Delete a scan if fully finished. """
399
400			if self.get_status(scan_id) == ScanStatus.RUNNING:
401			return False
402			self.scans_table.pop(scan_id)
403			if len(self.scans_table) == 0:
404			del self.data_manager
405			self.data_manager = None
406			return True
407
408
409			class ResultType(object):
410
411			""" Various scan results types values. """
412
413			ALARM = 0
414			LOG = 1
415			ERROR = 2
416			HOST_DETAIL = 3
417
418			@classmethod
419			def get_str(cls, result_type):
420			""" Return string name of a result type. """
421			if result_type == cls.ALARM:
422			return "Alarm"
423			elif result_type == cls.LOG:
424			return "Log Message"
425			elif result_type == cls.ERROR:
426			return "Error Message"
427			elif result_type == cls.HOST_DETAIL:
428			return "Host Detail"
429			else:
430			assert False, "Erroneous result type {0}.".format(result_type)
431
432			@classmethod
433			def get_type(cls, result_name):
434			""" Return string name of a result type. """
435			if result_name == "Alarm":
436			return cls.ALARM
437			elif result_name == "Log Message":
438			return cls.LOG
439			elif result_name == "Error Message":
440			return cls.ERROR
441			elif result_name == "Host Detail":
442			return cls.HOST_DETAIL
443			else:
444			assert False, "Erroneous result name {0}.".format(result_name)
445
446
447			def valid_uuid(value):
448			""" Check if value is a valid UUID. """
449
450			try:
451			uuid.UUID(value, version=4)
452			return True
453			except (TypeError, ValueError, AttributeError):
454			return False
455
456
457			def go_to_background():
458			""" Daemonize the running process. """
459			try:
460			if os.fork():
461			sys.exit()
462			except OSError as errmsg:
463			LOGGER.error('Fork failed: %s', errmsg)
464			sys.exit(1)
465
466
467			def create_pid(pidfile):
468			""" Check if there is an already running daemon and creates the pid file.
469			Otherwise gives an error. """
470
471			pid = str(os.getpid())
472
473			if Path(pidfile).is_file():
474			LOGGER.error("There is an already running process.")
475			return False
476
477			try:
478			with open(pidfile, 'w') as f:
479			f.write(pid)
480			except (FileNotFoundError, PermissionError) as e:
481			msg = "Failed to create pid file %s. %s" % (os.path.dirname(pidfile), e)
482			LOGGER.error(msg)
483			return False
484
485			return True
486
487
488			def remove_pidfile(pidfile, signum=None, frame=None):
489			""" Removes the pidfile before ending the daemon. """
490			pidpath = Path(pidfile)
491			if pidpath.is_file():
492			with pidpath.open() as f:
493			if int(f.read()) == os.getpid():
494			LOGGER.debug("Finishing daemon process")
495			pidpath.unlink()
496			sys.exit()
497

greenbone / ospd

Pull Request — master (#182)

ScanCollection.simplify_exclude_host_list() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like