ospd.scan.ScanCollection.get_hosts_finished() - Code Metrics - Inspection of "Cleanup #3" - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#205)

unknown

created 2020-02-13 11:11 UTC

ospd.scan.ScanCollection.get_hosts_finished() A

↳ Parent: ospd.scan

Complexity

Conditions

Size

Total Lines	10
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	2
dl	0
loc	10
rs	10
c	0
b	0
f	0

# Copyright (C) 2020 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.

import logging
import multiprocessing
import time
import uuid

from collections import OrderedDict
from enum import Enum
from typing import List, Any, Dict, Iterator, Optional

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:

    """ 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: Dict = None,
    ) -> 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) -> Dict:
        """ Get a scan's vts. """

        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


1			# Copyright (C) 2020 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			import logging
20			import multiprocessing
21			import time
22			import uuid
23
24			from collections import OrderedDict
25			from enum import Enum
26			from typing import List, Any, Dict, Iterator, Optional
27
28			from ospd.network import target_str_to_list
29
30			LOGGER = logging.getLogger(__name__)
31
32
33			class ScanStatus(Enum):
34			"""Scan status. """
35
36			INIT = 0
37			RUNNING = 1
38			STOPPED = 2
39			FINISHED = 3
40
41
42			class ScanCollection:
43
44			""" Scans collection, managing scans and results read and write, exposing
45			only needed information.
46
47			Each scan has meta-information such as scan ID, current progress (from 0 to
48			100), start time, end time, scan target and options and a list of results.
49
50			There are 4 types of results: Alarms, Logs, Errors and Host Details.
51
52			Todo:
53			- Better checking for Scan ID existence and handling otherwise.
54			- More data validation.
55			- Mutex access per table/scan_info.
56
57			"""
58
59			def __init__(self) -> None:
60			""" Initialize the Scan Collection. """
61
62			self.data_manager = (
63			None
64			) # type: Optional[multiprocessing.managers.SyncManager]
65			self.scans_table = dict() # type: Dict
66
67			def add_result(
68			self,
69			scan_id: str,
70			result_type: int,
71			host: str = '',
72			hostname: str = '',
73			name: str = '',
74			value: str = '',
75			port: str = '',
76			test_id: str = '',
77			severity: str = '',
78			qod: str = '',
79			) -> None:
80			""" Add a result to a scan in the table. """
81
82			assert scan_id
83			assert len(name) or len(value)
84
85			result = OrderedDict() # type: Dict
86			result['type'] = result_type
87			result['name'] = name
88			result['severity'] = severity
89			result['test_id'] = test_id
90			result['value'] = value
91			result['host'] = host
92			result['hostname'] = hostname
93			result['port'] = port
94			result['qod'] = qod
95			results = self.scans_table[scan_id]['results']
96			results.append(result)
97
98			# Set scan_info's results to propagate results to parent process.
99			self.scans_table[scan_id]['results'] = results
100
101			def remove_hosts_from_target_progress(
102			self, scan_id: str, target: str, hosts: List
103			) -> None:
104			"""Remove a list of hosts from the main scan progress table to avoid
105			the hosts to be included in the calculation of the scan progress"""
106			if not hosts:
107			return
108
109			targets = self.scans_table[scan_id]['target_progress']
110			for host in hosts:
111			if host in targets[target]:
112			del targets[target][host]
113
114			# Set scan_info's target_progress to propagate progresses
115			# to parent process.
116			self.scans_table[scan_id]['target_progress'] = targets
117
118			def set_progress(self, scan_id: str, progress: int) -> None:
119			""" Sets scan_id scan's progress. """
120
121			if progress > 0 and progress <= 100:
122			self.scans_table[scan_id]['progress'] = progress
123
124			if progress == 100:
125			self.scans_table[scan_id]['end_time'] = int(time.time())
126
127			def set_host_progress(
128			self, scan_id: str, target: str, host: str, progress: int
129			) -> None:
130			""" Sets scan_id scan's progress. """
131			if progress > 0 and progress <= 100:
132			targets = self.scans_table[scan_id]['target_progress']
133			targets[target][host] = progress
134			# Set scan_info's target_progress to propagate progresses
135			# to parent process.
136			self.scans_table[scan_id]['target_progress'] = targets
137
138			def set_host_finished(self, scan_id: str, target: str, host: str) -> None:
139			""" Add the host in a list of finished hosts """
140			finished_hosts = self.scans_table[scan_id]['finished_hosts']
141
142			if host not in finished_hosts[target]:
143			finished_hosts[target].append(host)
144
145			self.scans_table[scan_id]['finished_hosts'] = finished_hosts
146
147			def get_hosts_unfinished(self, scan_id: str) -> List[Any]:
148			""" Get a list of unfinished hosts."""
149
150			unfinished_hosts = list() # type: List
151
152			for target in self.scans_table[scan_id]['finished_hosts']:
153			unfinished_hosts.extend(target_str_to_list(target))
154
155			for target in self.scans_table[scan_id]['finished_hosts']:
156			for host in self.scans_table[scan_id]['finished_hosts'][target]:
157			unfinished_hosts.remove(host)
158
159			return unfinished_hosts
160
161			def get_hosts_finished(self, scan_id: str) -> List:
162			""" Get a list of finished hosts."""
163
164			finished_hosts = list() # type: List
165			for target in self.scans_table[scan_id]['finished_hosts']:
166			finished_hosts.extend(
167			self.scans_table[scan_id]['finished_hosts'].get(target)
168			)
169
170			return finished_hosts
171
172			def results_iterator(
173			self, scan_id: str, pop_res: bool = False, max_res: int = None
174			) -> Iterator[Any]:
175			""" Returns an iterator over scan_id scan's results. If pop_res is True,
176			it removed the fetched results from the list.
177
178			If max_res is None, return all the results.
179			Otherwise, if max_res = N > 0 return N as maximum number of results.
180
181			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 = None,
247			options: Optional[Dict] = None,
248			vts: Dict = None,
249			) -> str:
250			""" Creates a new scan with provided scan information. """
251
252			if not targets:
253			targets = []
254
255			if self.data_manager is None:
256			self.data_manager = multiprocessing.Manager()
257
258			# Check if it is possible to resume task. To avoid to resume, the
259			# scan must be deleted from the scans_table.
260			if (
261			scan_id
262			and self.id_exists(scan_id)
263			and (self.get_status(scan_id) == ScanStatus.STOPPED)
264			):
265			self.scans_table[scan_id]['end_time'] = 0
266
267			return self.resume_scan(scan_id, options)
268
269			if not options:
270			options = dict()
271
272			scan_info = self.data_manager.dict() # type: Dict
273			scan_info['results'] = list()
274			scan_info['finished_hosts'] = dict(
275			[[target, []] for target, _, _, _, _, _ in targets]
276			)
277			scan_info['progress'] = 0
278			scan_info['target_progress'] = dict(
279			[[target, {}] for target, _, _, _, _, _ in targets]
280			)
281			scan_info['targets'] = targets
282			scan_info['vts'] = vts
283			scan_info['options'] = options
284			scan_info['start_time'] = int(time.time())
285			scan_info['end_time'] = 0
286			scan_info['status'] = ScanStatus.INIT
287
288			if scan_id is None or scan_id == '':
289			scan_id = str(uuid.uuid4())
290
291			scan_info['scan_id'] = scan_id
292
293			self.scans_table[scan_id] = scan_info
294			return scan_id
295
296			def set_status(self, scan_id: str, status: ScanStatus) -> None:
297			""" Sets scan_id scan's status. """
298			self.scans_table[scan_id]['status'] = status
299			if status == ScanStatus.STOPPED:
300			self.scans_table[scan_id]['end_time'] = int(time.time())
301
302			def get_status(self, scan_id: str) -> ScanStatus:
303			""" Get scan_id scans's status."""
304
305			return self.scans_table[scan_id]['status']
306
307			def get_options(self, scan_id: str) -> Dict:
308			""" Get scan_id scan's options list. """
309
310			return self.scans_table[scan_id]['options']
311
312			def set_option(self, scan_id, name: str, value: Any) -> None:
313			""" Set a scan_id scan's name option to value. """
314
315			self.scans_table[scan_id]['options'][name] = value
316
317			def get_progress(self, scan_id: str) -> int:
318			""" Get a scan's current progress value. """
319
320			return self.scans_table[scan_id]['progress']
321
322			def simplify_exclude_host_list(
323			self, scan_id: str, target: Any
324			) -> List[Any]:
325			""" Remove from exclude_hosts the received hosts in the finished_hosts
326			list sent by the client.
327			The finished hosts are sent also as exclude hosts for backward
328			compatibility purposses.
329			"""
330
331			exc_hosts_list = target_str_to_list(
332			self.get_exclude_hosts(scan_id, target)
333			)
334
335			finished_hosts_list = target_str_to_list(
336			self.get_finished_hosts(scan_id, target)
337			)
338
339			if finished_hosts_list and exc_hosts_list:
340			for finished in finished_hosts_list:
341			if finished in exc_hosts_list:
342			exc_hosts_list.remove(finished)
343
344			return exc_hosts_list
345
346			def get_target_progress(self, scan_id: str, target: str) -> float:
347			""" Get a target's current progress value.
348			The value is calculated with the progress of each single host
349			in the target."""
350
351			total_hosts = len(target_str_to_list(target))
352			exc_hosts_list = self.simplify_exclude_host_list(scan_id, target)
353			exc_hosts = len(exc_hosts_list) if exc_hosts_list else 0
354			host_progresses = self.scans_table[scan_id]['target_progress'].get(
355			target
356			)
357
358			try:
359			t_prog = sum(host_progresses.values()) / (
360			total_hosts - exc_hosts
361			) # type: float
362			except ZeroDivisionError:
363			LOGGER.error(
364			"Zero division error in %s", self.get_target_progress.__name__
365			)
366			raise
367
368			return t_prog
369
370			def get_start_time(self, scan_id: str) -> str:
371			""" Get a scan's start time. """
372
373			return self.scans_table[scan_id]['start_time']
374
375			def get_end_time(self, scan_id: str) -> str:
376			""" Get a scan's end time. """
377
378			return self.scans_table[scan_id]['end_time']
379
380			def get_target_list(self, scan_id: str) -> List:
381			""" Get a scan's target list. """
382
383			target_list = []
384			for target, _, _, _, _, _ in self.scans_table[scan_id]['targets']:
385			target_list.append(target)
386
387			return target_list
388
389			def get_ports(self, scan_id: str, target: str):
390			""" Get a scan's ports list. If a target is specified
391			it will return the corresponding port for it. If not,
392			it returns the port item of the first nested list in
393			the target's list.
394			"""
395			if target:
396			for item in self.scans_table[scan_id]['targets']:
397			if target == item[0]:
398			return item[1]
399
400			return self.scans_table[scan_id]['targets'][0][1]
401
402			def get_exclude_hosts(self, scan_id: str, target: str):
403			""" Get an exclude host list for a given target.
404			"""
405			if target:
406			for item in self.scans_table[scan_id]['targets']:
407			if target == item[0]:
408			return item[3]
409
410			def get_finished_hosts(self, scan_id: str, target: str):
411			""" Get the finished host list sent by the client for a given target.
412			"""
413			if target:
414			for item in self.scans_table[scan_id]['targets']:
415			if target == item[0]:
416			return item[4]
417
418			def get_credentials(self, scan_id: str, target: str):
419			""" Get a scan's credential list. It return dictionary with
420			the corresponding credential for a given target.
421			"""
422			if target:
423			for item in self.scans_table[scan_id]['targets']:
424			if target == item[0]:
425			return item[2]
426
427			def get_target_options(self, scan_id: str, target: str):
428			""" Get a scan's target option dictionary.
429			It return dictionary with the corresponding options for
430			a given target.
431			"""
432			if target:
433			for item in self.scans_table[scan_id]['targets']:
434			if target == item[0]:
435			return item[5]
436
437			def get_vts(self, scan_id: str) -> Dict:
438			""" Get a scan's vts. """
439
440			return self.scans_table[scan_id]['vts']
441
442			def id_exists(self, scan_id: str) -> bool:
443			""" Check whether a scan exists in the table. """
444
445			return self.scans_table.get(scan_id) is not None
446
447			def delete_scan(self, scan_id: str) -> bool:
448			""" Delete a scan if fully finished. """
449
450			if self.get_status(scan_id) == ScanStatus.RUNNING:
451			return False
452
453			self.scans_table.pop(scan_id)
454
455			if len(self.scans_table) == 0:
456			del self.data_manager
457			self.data_manager = None
458
459			return True
460

greenbone / ospd

Pull Request — master (#205)

ospd.scan.ScanCollection.get_hosts_finished() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like