ospd.misc.ScanCollection.remove_hosts_from_target_progress() - Code Metrics - Inspection of "Extend OSP with finished_hosts to improve resume t..." - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#174)

by Juan José

created 2019-11-08 08:07 UTC

ScanCollection.remove_hosts_from_target_progress() A

↳ Parent: ospd.misc

Complexity

Conditions

Size

Total Lines	14
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	4
eloc	8
nop	4
dl	0
loc	14
rs	10
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']
        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 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 = target_str_to_list(
            self.get_exclude_hosts(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_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			finished_hosts[target].append(host)
140			self.scans_table[scan_id]['finished_hosts'] = finished_hosts
141
142			def get_hosts_unfinished(self, scan_id):
143			""" Get a list of unfinished hosts."""
144
145			unfinished_hosts = list()
146			for target in self.scans_table[scan_id]['finished_hosts']:
147			unfinished_hosts.extend(target_str_to_list(target))
148			for target in self.scans_table[scan_id]['finished_hosts']:
149			for host in self.scans_table[scan_id]['finished_hosts'][target]:
150			unfinished_hosts.remove(host)
151
152			return unfinished_hosts
153
154			def get_hosts_finished(self, scan_id):
155			""" Get a list of finished hosts."""
156
157			finished_hosts = list()
158			for target in self.scans_table[scan_id]['finished_hosts']:
159			finished_hosts.extend(
160			self.scans_table[scan_id]['finished_hosts'].get(target)
161			)
162
163			return finished_hosts
164
165			def results_iterator(self, scan_id, pop_res):
166			""" Returns an iterator over scan_id scan's results. If pop_res is True,
167			it removed the fetched results from the list.
168			"""
169			if pop_res:
170			result_aux = self.scans_table[scan_id]['results']
171			self.scans_table[scan_id]['results'] = list()
172			return iter(result_aux)
173
174			return iter(self.scans_table[scan_id]['results'])
175
176			def ids_iterator(self):
177			""" Returns an iterator over the collection's scan IDS. """
178
179			return iter(self.scans_table.keys())
180
181			def remove_single_result(self, scan_id, result):
182			"""Removes a single result from the result list in scan_table.
183
184			Parameters:
185			scan_id (uuid): Scan ID to identify the scan process to be resumed.
186			result (dict): The result to be removed from the results list.
187			"""
188			results = self.scans_table[scan_id]['results']
189			results.remove(result)
190			self.scans_table[scan_id]['results'] = results
191
192			def del_results_for_stopped_hosts(self, scan_id):
193			""" Remove results from the result table for those host
194			"""
195			unfinished_hosts = self.get_hosts_unfinished(scan_id)
196			for result in self.results_iterator(scan_id, False):
197			if result['host'] in unfinished_hosts:
198			self.remove_single_result(scan_id, result)
199
200			def resume_scan(self, scan_id, options):
201			""" Reset the scan status in the scan_table to INIT.
202			Also, overwrite the options, because a resume task cmd
203			can add some new option. E.g. exclude hosts list.
204			Parameters:
205			scan_id (uuid): Scan ID to identify the scan process to be resumed.
206			options (dict): Options for the scan to be resumed. This options
207			are not added to the already existent ones.
208			The old ones are removed
209
210			Return:
211			Scan ID which identifies the current scan.
212			"""
213			self.scans_table[scan_id]['status'] = ScanStatus.INIT
214			if options:
215			self.scans_table[scan_id]['options'] = options
216
217			self.del_results_for_stopped_hosts(scan_id)
218
219			return scan_id
220
221			def create_scan(self, scan_id='', targets='', options=None, vts=''):
222			""" Creates a new scan with provided scan information. """
223
224			if self.data_manager is None:
225			self.data_manager = multiprocessing.Manager()
226
227			# Check if it is possible to resume task. To avoid to resume, the
228			# scan must be deleted from the scans_table.
229			if (
230			scan_id
231			and self.id_exists(scan_id)
232			and (self.get_status(scan_id) == ScanStatus.STOPPED)
233			):
234			self.scans_table[scan_id]['end_time'] = 0
235
236			return self.resume_scan(scan_id, options)
237
238			if not options:
239			options = dict()
240			scan_info = self.data_manager.dict()
241			scan_info['results'] = list()
242			scan_info['finished_hosts'] = dict(
243			[[target, []] for target, _, _, _, _ in targets]
244			)
245			scan_info['progress'] = 0
246			scan_info['target_progress'] = dict(
247			[[target, {}] for target, _, _, _, _ in targets]
248			)
249			scan_info['targets'] = targets
250			scan_info['vts'] = vts
251			scan_info['options'] = options
252			scan_info['start_time'] = int(time.time())
253			scan_info['end_time'] = 0
254			scan_info['status'] = ScanStatus.INIT
255			if scan_id is None or scan_id == '':
256			scan_id = str(uuid.uuid4())
257			scan_info['scan_id'] = scan_id
258			self.scans_table[scan_id] = scan_info
259			return scan_id
260
261			def set_status(self, scan_id, status):
262			""" Sets scan_id scan's status. """
263			self.scans_table[scan_id]['status'] = status
264			if status == ScanStatus.STOPPED:
265			self.scans_table[scan_id]['end_time'] = int(time.time())
266
267			def get_status(self, scan_id):
268			""" Get scan_id scans's status."""
269
270			return self.scans_table[scan_id]['status']
271
272			def get_options(self, scan_id):
273			""" Get scan_id scan's options list. """
274
275			return self.scans_table[scan_id]['options']
276
277			def set_option(self, scan_id, name, value):
278			""" Set a scan_id scan's name option to value. """
279
280			self.scans_table[scan_id]['options'][name] = value
281
282			def get_progress(self, scan_id):
283			""" Get a scan's current progress value. """
284
285			return self.scans_table[scan_id]['progress']
286
287			def get_target_progress(self, scan_id, target):
288			""" Get a target's current progress value.
289			The value is calculated with the progress of each single host
290			in the target."""
291
292			total_hosts = len(target_str_to_list(target))
293			exc_hosts_list = target_str_to_list(
294			self.get_exclude_hosts(scan_id, target)
295			)
296			exc_hosts = len(exc_hosts_list) if exc_hosts_list else 0
297			host_progresses = self.scans_table[scan_id]['target_progress'].get(
298			target
299			)
300			try:
301			t_prog = sum(host_progresses.values()) / (total_hosts - exc_hosts)
302			except ZeroDivisionError:
303			LOGGER.error(
304			"Zero division error in %s", self.get_target_progress.__name__
305			)
306			raise
307			return t_prog
308
309			def get_start_time(self, scan_id):
310			""" Get a scan's start time. """
311
312			return self.scans_table[scan_id]['start_time']
313
314			def get_end_time(self, scan_id):
315			""" Get a scan's end time. """
316
317			return self.scans_table[scan_id]['end_time']
318
319			def get_target_list(self, scan_id):
320			""" Get a scan's target list. """
321
322			target_list = []
323			for target, _, _, _, _ in self.scans_table[scan_id]['targets']:
324			target_list.append(target)
325			return target_list
326
327			def get_ports(self, scan_id, target):
328			""" Get a scan's ports list. If a target is specified
329			it will return the corresponding port for it. If not,
330			it returns the port item of the first nested list in
331			the target's list.
332			"""
333			if target:
334			for item in self.scans_table[scan_id]['targets']:
335			if target == item[0]:
336			return item[1]
337
338			return self.scans_table[scan_id]['targets'][0][1]
339
340			def get_exclude_hosts(self, scan_id, target):
341			""" Get an exclude host list for a given target.
342			"""
343			if target:
344			for item in self.scans_table[scan_id]['targets']:
345			if target == item[0]:
346			return item[3]
347
348			def get_credentials(self, scan_id, target):
349			""" Get a scan's credential list. It return dictionary with
350			the corresponding credential for a given target.
351			"""
352			if target:
353			for item in self.scans_table[scan_id]['targets']:
354			if target == item[0]:
355			return item[2]
356
357			def get_vts(self, scan_id):
358			""" Get a scan's vts list. """
359
360			return self.scans_table[scan_id]['vts']
361
362			def id_exists(self, scan_id):
363			""" Check whether a scan exists in the table. """
364
365			return self.scans_table.get(scan_id) is not None
366
367			def delete_scan(self, scan_id):
368			""" Delete a scan if fully finished. """
369
370			if self.get_status(scan_id) == ScanStatus.RUNNING:
371			return False
372			self.scans_table.pop(scan_id)
373			if len(self.scans_table) == 0:
374			del self.data_manager
375			self.data_manager = None
376			return True
377
378
379			class ResultType(object):
380
381			""" Various scan results types values. """
382
383			ALARM = 0
384			LOG = 1
385			ERROR = 2
386			HOST_DETAIL = 3
387
388			@classmethod
389			def get_str(cls, result_type):
390			""" Return string name of a result type. """
391			if result_type == cls.ALARM:
392			return "Alarm"
393			elif result_type == cls.LOG:
394			return "Log Message"
395			elif result_type == cls.ERROR:
396			return "Error Message"
397			elif result_type == cls.HOST_DETAIL:
398			return "Host Detail"
399			else:
400			assert False, "Erroneous result type {0}.".format(result_type)
401
402			@classmethod
403			def get_type(cls, result_name):
404			""" Return string name of a result type. """
405			if result_name == "Alarm":
406			return cls.ALARM
407			elif result_name == "Log Message":
408			return cls.LOG
409			elif result_name == "Error Message":
410			return cls.ERROR
411			elif result_name == "Host Detail":
412			return cls.HOST_DETAIL
413			else:
414			assert False, "Erroneous result name {0}.".format(result_name)
415
416
417			def valid_uuid(value):
418			""" Check if value is a valid UUID. """
419
420			try:
421			uuid.UUID(value, version=4)
422			return True
423			except (TypeError, ValueError, AttributeError):
424			return False
425
426
427			def go_to_background():
428			""" Daemonize the running process. """
429			try:
430			if os.fork():
431			sys.exit()
432			except OSError as errmsg:
433			LOGGER.error('Fork failed: %s', errmsg)
434			sys.exit(1)
435
436
437			def create_pid(pidfile):
438			""" Check if there is an already running daemon and creates the pid file.
439			Otherwise gives an error. """
440
441			pid = str(os.getpid())
442
443			if Path(pidfile).is_file():
444			LOGGER.error("There is an already running process.")
445			return False
446
447			try:
448			with open(pidfile, 'w') as f:
449			f.write(pid)
450			except (FileNotFoundError, PermissionError) as e:
451			msg = "Failed to create pid file %s. %s" % (os.path.dirname(pidfile), e)
452			LOGGER.error(msg)
453			return False
454
455			return True
456
457
458			def remove_pidfile(pidfile, signum=None, frame=None):
459			""" Removes the pidfile before ending the daemon. """
460			pidpath = Path(pidfile)
461			if pidpath.is_file():
462			with pidpath.open() as f:
463			if int(f.read()) == os.getpid():
464			LOGGER.debug("Finishing daemon process")
465			pidpath.unlink()
466			sys.exit()
467

greenbone / ospd

Pull Request — master (#174)

ScanCollection.remove_hosts_from_target_progress() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like