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

Passed

Pull Request — master (#198)

unknown

created 2020-02-07 07:36 UTC

ospd.misc.ScanCollection.create_scan() C

↳ Parent: ospd.misc

Complexity

Conditions

Size

Total Lines	48
Code Lines	36

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
eloc	36
nop	5
dl	0
loc	48
rs	6.6666
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, Callable, Iterable
from enum import Enum
from collections import OrderedDict
from pathlib import Path

from ospd.network import target_str_to_list

LOGGER = logging.getLogger(__name__)


def create_process(
    func: Callable, *, args: Iterable[Any] = None
) -> multiprocessing.Process:
    return multiprocessing.Process(target=func, args=args)


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 = None,
        options: Optional[Dict] = None,
        vts: str = '',
    ) -> str:
        """ Creates a new scan with provided scan information. """

        if not targets:
            targets = []

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

greenbone / ospd

Pull Request — master (#198)

ospd.misc.ScanCollection.create_scan() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like