ospd.misc.ScanCollection.get_target_options() - Code Metrics - Inspection of "Extend osp to accept target options." - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#194)

by Juan José

created 2020-01-17 13:22 UTC

ospd.misc.ScanCollection.get_target_options() A

↳ Parent: ospd.misc

Complexity

Conditions

Size

Total Lines	9
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	4
eloc	5
nop	3
dl	0
loc	9
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 typing import List, Any, Dict, Iterator, Optional
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) -> None:
        """ Initialize the Scan Collection. """

        self.data_manager = (
            None
        )  # type: Optional[multiprocessing.managers.SyncManager]
        self.scans_table = dict()  # type: Dict

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

        assert scan_id
        assert len(name) or len(value)
        result = OrderedDict()  # type: 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 remove_hosts_from_target_progress(
        self, scan_id: str, target: str, hosts: List
    ) -> None:
        """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: str, progress: int) -> None:
        """ 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: str, target: str, host: str, progress: int
    ) -> None:
        """ 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: str, target: str, host: str) -> None:
        """ 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: str) -> List[Any]:
        """ Get a list of unfinished hosts."""

        unfinished_hosts = list()  # type: 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: str) -> List:
        """ Get a list of finished hosts."""

        finished_hosts = list()  # type: 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: str, pop_res: bool = False, max_res: int = None
    ) -> Iterator[Any]:
        """ Returns an iterator over scan_id scan's results. If pop_res is True,
        it removed the fetched results from the list.
        If max_res is None, return all the results. Otherwise, if max_res = N > 0
        return N as maximum number of results. max_res works only together with pop_results.
        """
        if pop_res and max_res:
            result_aux = self.scans_table[scan_id]['results']
            self.scans_table[scan_id]['results'] = result_aux[max_res:]
            return iter(result_aux[:max_res])
        elif 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) -> Iterator[str]:
        """ Returns an iterator over the collection's scan IDS. """

        return iter(self.scans_table.keys())

    def remove_single_result(
        self, scan_id: str, result: Dict[str, str]
    ) -> None:
        """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: str) -> None:
        """ 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, pop_res=False, max_res=None
        ):
            if result['host'] in unfinished_hosts:
                self.remove_single_result(scan_id, result)

    def resume_scan(self, scan_id: str, options: Optional[Dict]) -> str:
        """ 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: str = '',
        targets: List = [],
        options: Optional[Dict] = None,
        vts: str = '',
    ) -> str:
        """ 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()  # type: 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: str, status: ScanStatus) -> None:
        """ 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: str) -> ScanStatus:
        """ Get scan_id scans's status."""

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

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

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

    def set_option(self, scan_id, name: str, value: Any) -> None:
        """ Set a scan_id scan's name option to value. """

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

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

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

    def simplify_exclude_host_list(
        self, scan_id: str, target: Any
    ) -> List[Any]:
        """ 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: str, target: str) -> float:
        """ 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
            )  # type: float
        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: str) -> str:
        """ Get a scan's start time. """

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

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

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

    def get_target_list(self, scan_id: str) -> List:
        """ 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: str, target: str):
        """ 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: str, target: str):
        """ 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: str, target: str):
        """ 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: str, target: str):
        """ 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_target_options(self, scan_id: str, target: str):
        """ Get a scan's target option dictionary.
        It return dictionary with the corresponding options for
        a given target.
        """
        if target:
            for item in self.scans_table[scan_id]['targets']:
                if target == item[0]:
                    return item[5]

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

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

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

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

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

greenbone / ospd

Pull Request — master (#194)

ospd.misc.ScanCollection.get_target_options() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like