ospd.scan.ScanCollection.get_start_time() - Code Metrics - Inspection of "Improve handling of dead hosts and improve scan pr..." - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#266)

by Juan José

created 2020-05-11 12:31 UTC

ospd.scan.ScanCollection.get_start_time() A

↳ Parent: ospd.scan

Complexity

Conditions

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# Copyright (C) 2014-2020 Greenbone Networks GmbH
#
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

import logging
import multiprocessing
import time
import uuid

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

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 add_result_list(
        self, scan_id: str, result_list: Iterable[Dict[str, str]]
    ) -> None:
        """
        Add a batch of results to the result's table for the corresponding
        scan_id
        """
        results = self.scans_table[scan_id]['results']
        results.extend(result_list)

        # 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, 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

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

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

    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, host_progress_batch: Dict[str, int]
    ) -> None:
        """ Sets scan_id scan's progress. """
        host_progresses = self.scans_table[scan_id].get('target_progress')
        host_progresses.update(host_progress_batch)

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

    def set_host_finished(self, scan_id: str, hosts: List[str]) -> None:
        """ Increase the amount of finished hosts which were alive."""

        total_finished = len(hosts)
        count_alive = (
            self.scans_table[scan_id].get('count_alive') + total_finished
        )
        self.scans_table[scan_id]['count_alive'] = count_alive

    def set_host_dead(self, scan_id: str, hosts: List[str]) -> None:
        """ Increase the amount of dead hosts. """

        total_dead = len(hosts)
        count_dead = self.scans_table[scan_id].get('count_dead') + total_dead
        self.scans_table[scan_id]['count_dead'] = count_dead

    def set_amount_dead_hosts(self, scan_id: str, total_dead: int) -> None:
        """ Increase the amount of dead hosts. """

        count_dead = self.scans_table[scan_id].get('count_dead') + total_dead
        self.scans_table[scan_id]['count_dead'] = count_dead

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

        if not target:
            target = {}

        if self.data_manager is None:
            self.data_manager = multiprocessing.Manager()

        if not options:
            options = dict()

        scan_info = self.data_manager.dict()  # type: Dict
        scan_info['results'] = list()
        scan_info['progress'] = 0
        scan_info['target_progress'] = dict()
        scan_info['count_alive'] = 0
        scan_info['count_dead'] = 0
        scan_info['target'] = target
        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 get_count_dead(self, scan_id: str) -> int:
        """ Get a scan's current dead host count. """

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

    def get_count_alive(self, scan_id: str) -> int:
        """ Get a scan's current dead host count. """

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

    def get_current_target_progress(self, scan_id: str) -> Dict[str, int]:
        """ Get a scan's current dead host count. """

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

    def simplify_exclude_host_count(self, scan_id: str) -> int:
        """ 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.

        Return:
            Count of excluded host.
        """

        exc_hosts_list = target_str_to_list(self.get_exclude_hosts(scan_id))

        finished_hosts_list = target_str_to_list(
            self.get_finished_hosts(scan_id)
        )

        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 len(exc_hosts_list) if exc_hosts_list else 0

    def calculate_target_progress(self, scan_id: 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 = self.get_host_count(scan_id)
        exc_hosts = self.simplify_exclude_host_count(scan_id)
        count_alive = self.get_count_alive(scan_id)
        count_dead = self.get_count_dead(scan_id)
        host_progresses = self.scans_table[scan_id].get('target_progress')

        try:
            t_prog = (sum(host_progresses.values()) + 100 * count_alive) / (
                total_hosts - exc_hosts - count_dead
            )  # type: float
        except ZeroDivisionError:
            LOGGER.error(
                "Zero division error in %s",
                self.calculate_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_host_list(self, scan_id: str) -> Dict:
        """ Get a scan's host list. """

        return self.scans_table[scan_id]['target'].get('hosts')

    def get_host_count(self, scan_id: str) -> int:
        """ Get total host count in the target. """
        host = self.get_host_list(scan_id)
        total_hosts = len(target_str_to_list(host))

        return total_hosts

    def get_ports(self, scan_id: str):
        """ Get a scan's ports list.
        """
        return self.scans_table[scan_id]['target'].get('ports')

    def get_exclude_hosts(self, scan_id: str):
        """ Get an exclude host list for a given target.
        """
        return self.scans_table[scan_id]['target'].get('exclude_hosts')

    def get_finished_hosts(self, scan_id: str):
        """ Get the finished host list sent by the client for a given target.
        """
        return self.scans_table[scan_id]['target'].get('finished_hosts')

    def get_credentials(self, scan_id: str):
        """ Get a scan's credential list. It return dictionary with
        the corresponding credential for a given target.
        """
        return self.scans_table[scan_id]['target'].get('credentials')

    def get_target_options(self, scan_id: str):
        """ Get a scan's target option dictionary.
        It return dictionary with the corresponding options for
        a given target.
        """
        return self.scans_table[scan_id]['target'].get('options')

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

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

    def release_vts_list(self, scan_id: str) -> None:
        """ Release the memory used for the vts list. """

        scan_data = self.scans_table.get(scan_id)
        if scan_data and 'vts' in scan_data:
            del scan_data['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) 2014-2020 Greenbone Networks GmbH
2			#
3			# SPDX-License-Identifier: AGPL-3.0-or-later
4			#
5			# This program is free software: you can redistribute it and/or modify
6			# it under the terms of the GNU Affero General Public License as
7			# published by the Free Software Foundation, either version 3 of the
8			# 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 Affero General Public License for more details.
14			#
15			# You should have received a copy of the GNU Affero General Public License
16			# along with this program. If not, see <http://www.gnu.org/licenses/>.
17
18			import logging
19			import multiprocessing
20			import time
21			import uuid
22
23			from collections import OrderedDict
24			from enum import Enum
25			from typing import List, Any, Dict, Iterator, Optional, Iterable
26
27			from ospd.network import target_str_to_list
28
29			LOGGER = logging.getLogger(__name__)
30
31
32			class ScanStatus(Enum):
33			"""Scan status. """
34
35			INIT = 0
36			RUNNING = 1
37			STOPPED = 2
38			FINISHED = 3
39
40
41			class ScanCollection:
42
43			""" Scans collection, managing scans and results read and write, exposing
44			only needed information.
45
46			Each scan has meta-information such as scan ID, current progress (from 0 to
47			100), start time, end time, scan target and options and a list of results.
48
49			There are 4 types of results: Alarms, Logs, Errors and Host Details.
50
51			Todo:
52			- Better checking for Scan ID existence and handling otherwise.
53			- More data validation.
54			- Mutex access per table/scan_info.
55
56			"""
57
58			def __init__(self) -> None:
59			""" Initialize the Scan Collection. """
60
61			self.data_manager = (
62			None
63			) # type: Optional[multiprocessing.managers.SyncManager]
64			self.scans_table = dict() # type: Dict
65
66			def add_result(
67			self,
68			scan_id: str,
69			result_type: int,
70			host: str = '',
71			hostname: str = '',
72			name: str = '',
73			value: str = '',
74			port: str = '',
75			test_id: str = '',
76			severity: str = '',
77			qod: str = '',
78			) -> None:
79			""" Add a result to a scan in the table. """
80
81			assert scan_id
82			assert len(name) or len(value)
83
84			result = OrderedDict() # type: Dict
85			result['type'] = result_type
86			result['name'] = name
87			result['severity'] = severity
88			result['test_id'] = test_id
89			result['value'] = value
90			result['host'] = host
91			result['hostname'] = hostname
92			result['port'] = port
93			result['qod'] = qod
94			results = self.scans_table[scan_id]['results']
95			results.append(result)
96
97			# Set scan_info's results to propagate results to parent process.
98			self.scans_table[scan_id]['results'] = results
99
100			def add_result_list(
101			self, scan_id: str, result_list: Iterable[Dict[str, str]]
102			) -> None:
103			"""
104			Add a batch of results to the result's table for the corresponding
105			scan_id
106			"""
107			results = self.scans_table[scan_id]['results']
108			results.extend(result_list)
109
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, 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			target = self.scans_table[scan_id].get('target_progress')
122			for host in hosts:
123			if host in target:
124			del 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'] = target
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
136			if progress == 100:
137			self.scans_table[scan_id]['end_time'] = int(time.time())
138
139			def set_host_progress(
140			self, scan_id: str, host_progress_batch: Dict[str, int]
141			) -> None:
142			""" Sets scan_id scan's progress. """
143			host_progresses = self.scans_table[scan_id].get('target_progress')
144			host_progresses.update(host_progress_batch)
145
146			# Set scan_info's target_progress to propagate progresses
147			# to parent process.
148			self.scans_table[scan_id]['target_progress'] = host_progresses
149
150			def set_host_finished(self, scan_id: str, hosts: List[str]) -> None:
151			""" Increase the amount of finished hosts which were alive."""
152
153			total_finished = len(hosts)
154			count_alive = (
155			self.scans_table[scan_id].get('count_alive') + total_finished
156			)
157			self.scans_table[scan_id]['count_alive'] = count_alive
158
159			def set_host_dead(self, scan_id: str, hosts: List[str]) -> None:
160			""" Increase the amount of dead hosts. """
161
162			total_dead = len(hosts)
163			count_dead = self.scans_table[scan_id].get('count_dead') + total_dead
164			self.scans_table[scan_id]['count_dead'] = count_dead
165
166			def set_amount_dead_hosts(self, scan_id: str, total_dead: int) -> None:
167			""" Increase the amount of dead hosts. """
168
169			count_dead = self.scans_table[scan_id].get('count_dead') + total_dead
170			self.scans_table[scan_id]['count_dead'] = count_dead
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 create_scan(
200			self,
201			scan_id: str = '',
202			target: Dict = None,
203			options: Optional[Dict] = None,
204			vts: Dict = None,
205			) -> str:
206			""" Creates a new scan with provided scan information. """
207
208			if not target:
209			target = {}
210
211			if self.data_manager is None:
212			self.data_manager = multiprocessing.Manager()
213
214			if not options:
215			options = dict()
216
217			scan_info = self.data_manager.dict() # type: Dict
218			scan_info['results'] = list()
219			scan_info['progress'] = 0
220			scan_info['target_progress'] = dict()
221			scan_info['count_alive'] = 0
222			scan_info['count_dead'] = 0
223			scan_info['target'] = target
224			scan_info['vts'] = vts
225			scan_info['options'] = options
226			scan_info['start_time'] = int(time.time())
227			scan_info['end_time'] = 0
228			scan_info['status'] = ScanStatus.INIT
229
230			if scan_id is None or scan_id == '':
231			scan_id = str(uuid.uuid4())
232
233			scan_info['scan_id'] = scan_id
234
235			self.scans_table[scan_id] = scan_info
236			return scan_id
237
238			def set_status(self, scan_id: str, status: ScanStatus) -> None:
239			""" Sets scan_id scan's status. """
240			self.scans_table[scan_id]['status'] = status
241			if status == ScanStatus.STOPPED:
242			self.scans_table[scan_id]['end_time'] = int(time.time())
243
244			def get_status(self, scan_id: str) -> ScanStatus:
245			""" Get scan_id scans's status."""
246
247			return self.scans_table[scan_id]['status']
248
249			def get_options(self, scan_id: str) -> Dict:
250			""" Get scan_id scan's options list. """
251
252			return self.scans_table[scan_id]['options']
253
254			def set_option(self, scan_id, name: str, value: Any) -> None:
255			""" Set a scan_id scan's name option to value. """
256
257			self.scans_table[scan_id]['options'][name] = value
258
259			def get_progress(self, scan_id: str) -> int:
260			""" Get a scan's current progress value. """
261
262			return self.scans_table[scan_id]['progress']
263
264			def get_count_dead(self, scan_id: str) -> int:
265			""" Get a scan's current dead host count. """
266
267			return self.scans_table[scan_id]['count_dead']
268
269			def get_count_alive(self, scan_id: str) -> int:
270			""" Get a scan's current dead host count. """
271
272			return self.scans_table[scan_id]['count_alive']
273
274			def get_current_target_progress(self, scan_id: str) -> Dict[str, int]:
275			""" Get a scan's current dead host count. """
276
277			return self.scans_table[scan_id]['target_progress']
278
279			def simplify_exclude_host_count(self, scan_id: str) -> int:
280			""" Remove from exclude_hosts the received hosts in the finished_hosts
281			list sent by the client.
282			The finished hosts are sent also as exclude hosts for backward
283			compatibility purposses.
284
285			Return:
286			Count of excluded host.
287			"""
288
289			exc_hosts_list = target_str_to_list(self.get_exclude_hosts(scan_id))
290
291			finished_hosts_list = target_str_to_list(
292			self.get_finished_hosts(scan_id)
293			)
294
295			if finished_hosts_list and exc_hosts_list:
296			for finished in finished_hosts_list:
297			if finished in exc_hosts_list:
298			exc_hosts_list.remove(finished)
299
300			return len(exc_hosts_list) if exc_hosts_list else 0
301
302			def calculate_target_progress(self, scan_id: str) -> float:
303			""" Get a target's current progress value.
304			The value is calculated with the progress of each single host
305			in the target."""
306
307			total_hosts = self.get_host_count(scan_id)
308			exc_hosts = self.simplify_exclude_host_count(scan_id)
309			count_alive = self.get_count_alive(scan_id)
310			count_dead = self.get_count_dead(scan_id)
311			host_progresses = self.scans_table[scan_id].get('target_progress')
312
313			try:
314			t_prog = (sum(host_progresses.values()) + 100 * count_alive) / (
315			total_hosts - exc_hosts - count_dead
316			) # type: float
317			except ZeroDivisionError:
318			LOGGER.error(
319			"Zero division error in %s",
320			self.calculate_target_progress.__name__,
321			)
322			raise
323
324			return t_prog
325
326			def get_start_time(self, scan_id: str) -> str:
327			""" Get a scan's start time. """
328
329			return self.scans_table[scan_id]['start_time']
330
331			def get_end_time(self, scan_id: str) -> str:
332			""" Get a scan's end time. """
333
334			return self.scans_table[scan_id]['end_time']
335
336			def get_host_list(self, scan_id: str) -> Dict:
337			""" Get a scan's host list. """
338
339			return self.scans_table[scan_id]['target'].get('hosts')
340
341			def get_host_count(self, scan_id: str) -> int:
342			""" Get total host count in the target. """
343			host = self.get_host_list(scan_id)
344			total_hosts = len(target_str_to_list(host))
345
346			return total_hosts
347
348			def get_ports(self, scan_id: str):
349			""" Get a scan's ports list.
350			"""
351			return self.scans_table[scan_id]['target'].get('ports')
352
353			def get_exclude_hosts(self, scan_id: str):
354			""" Get an exclude host list for a given target.
355			"""
356			return self.scans_table[scan_id]['target'].get('exclude_hosts')
357
358			def get_finished_hosts(self, scan_id: str):
359			""" Get the finished host list sent by the client for a given target.
360			"""
361			return self.scans_table[scan_id]['target'].get('finished_hosts')
362
363			def get_credentials(self, scan_id: str):
364			""" Get a scan's credential list. It return dictionary with
365			the corresponding credential for a given target.
366			"""
367			return self.scans_table[scan_id]['target'].get('credentials')
368
369			def get_target_options(self, scan_id: str):
370			""" Get a scan's target option dictionary.
371			It return dictionary with the corresponding options for
372			a given target.
373			"""
374			return self.scans_table[scan_id]['target'].get('options')
375
376			def get_vts(self, scan_id: str) -> Dict:
377			""" Get a scan's vts. """
378
379			return self.scans_table[scan_id]['vts']
380
381			def release_vts_list(self, scan_id: str) -> None:
382			""" Release the memory used for the vts list. """
383
384			scan_data = self.scans_table.get(scan_id)
385			if scan_data and 'vts' in scan_data:
386			del scan_data['vts']
387
388			def id_exists(self, scan_id: str) -> bool:
389			""" Check whether a scan exists in the table. """
390
391			return self.scans_table.get(scan_id) is not None
392
393			def delete_scan(self, scan_id: str) -> bool:
394			""" Delete a scan if fully finished. """
395
396			if self.get_status(scan_id) == ScanStatus.RUNNING:
397			return False
398
399			self.scans_table.pop(scan_id)
400
401			if len(self.scans_table) == 0:
402			del self.data_manager
403			self.data_manager = None
404
405			return True
406

greenbone / ospd

Pull Request — master (#266)

ospd.scan.ScanCollection.get_start_time() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like