ospd_openvas.db.OpenvasDB.set_single_item() - Code Metrics - Inspection of "Merge pull request #41 from jjnicola/api-cleanup2" - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( b2d3e0...9e273c )

by Juan José

created 2018-12-11 14:27 UTC

ospd_openvas.db.OpenvasDB.set_single_item() A

↳ Parent: ospd_openvas.db

Complexity

Conditions

Size

Total Lines	18
Code Lines	11

Duplication

Lines	18
Ratio	100 %

Importance

Changes

Metric	Value
cc	4
eloc	11
nop	4
dl	18
loc	18
rs	9.85
c	0
b	0
f	0

# -*- coding: utf-8 -*-
# Copyright (C) 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

""" Access management for redis-based OpenVAS Scanner Database."""

import redis
import subprocess

from ospd_openvas.errors import OSPDOpenvasError
from ospd_openvas.errors import RequiredArgument
from ospd.ospd import logger

SOCKET_TIMEOUT = 60  # in seconds
LIST_FIRST_POS = 0
LIST_LAST_POS = -1
LIST_ALL = 0

# Possible positions of nvt values in cache list.
NVT_META_FIELDS = [
    "NVT_FILENAME_POS",
    "NVT_REQUIRED_KEYS_POS",
    "NVT_MANDATORY_KEYS_POS",
    "NVT_EXCLUDED_KEYS_POS",
    "NVT_REQUIRED_UDP_PORTS_POS",
    "NVT_REQUIRED_PORTS_POS",
    "NVT_DEPENDENCIES_POS",
    "NVT_TAGS_POS",
    "NVT_CVES_POS",
    "NVT_BIDS_POS",
    "NVT_XREFS_POS",
    "NVT_CATEGORY_POS",
    "NVT_TIMEOUT_POS",
    "NVT_FAMILY_POS",
    "NVT_NAME_POS",
]


class OpenvasDB(object):

    """ Class to connect to redis, to perform queries, and to move
    from a KB to another."""
    # Name of the namespace usage bitmap in redis.
    DBINDEX_NAME = "GVM.__GlobalDBIndex"

    def __init__(self):
        # Path to the Redis socket.
        self.db_address = None

        self.max_dbindex = 0
        self.db_index = 0
        self.rediscontext = None

    @staticmethod
    def _parse_openvassd_db_address(result):
        """ Return the path to the redis socket.
        Arguments:
            result (bytes) Output of `openvassd -s`
        Return redis unix socket path.
        """
        path = None
        result = result.decode('ascii')
        for conf in result.split('\n'):
            if conf.find('db_address') == 0:
                path = conf.split('=')
                break

        if not path:
            raise OSPDOpenvasError('Redis Error: Not possible to '
                                   'find the path to the redis socket.')
        return path[1].strip()

    def get_db_connection(self):
        """ Retrieve the db address from openvassd config.
        """
        result = subprocess.check_output(
            ['openvassd', '-s'], stderr=subprocess.STDOUT)

        if result:
            path = self._parse_openvassd_db_address(result)

        self.db_address = path


    def max_db_index(self):
        """Set the number of databases have been configured into kbr struct.
        """
        ctx = self.kb_connect()
        resp = ctx.config_get('databases')

        if len(resp) == 1:
            self.max_dbindex = int(resp.get('databases'))
        else:
            raise OSPDOpenvasError('Redis Error: Not possible '
                                   'to get max_dbindex.')

    def set_redisctx(self, ctx):
        """ Set the current rediscontext.
        Arguments:
            ctx (object): Redis context to be set as default.
        """
        if not ctx:
            raise RequiredArgument('set_redisctx: A valid Redis context is '
                                   'required.')
        self.rediscontext = ctx

    def db_init(self):
        """ Set db_address and max_db_index. """
        self.get_db_connection()
        self.max_db_index()

    def try_database_index(self, ctx, kb):
        """ Check if a redis kb is already in use. If not, set it
        as in use and return.
        Arguments:
            ctx (object): Redis object connected to the kb with the
                DBINDEX_NAME key.
            kb (int): Kb number intended to be used.

        Return True if it is possible to use the kb. False if the given kb
            number is already in use.
        """
        _IN_USE = 1
        try:
            resp = ctx.hsetnx(self.DBINDEX_NAME, kb, _IN_USE)
        except:
            raise OSPDOpenvasError('Redis Error: Not possible '
                                   'to set %s.' % self.DBINDEX_NAME)

        if resp == 1:
            return True
        return False

    def kb_connect(self, dbnum=0):
        """ Connect to redis to the given database or to the default db 0 .

        Arguments:
            dbnum (int, optional): The db number to connect to.

        Return a redis context on success.
        """
        self.get_db_connection()

        try:
            ctx = redis.Redis(unix_socket_path=self.db_address,
                              db=dbnum,
                              socket_timeout=SOCKET_TIMEOUT, charset="latin-1",
                              decode_responses=True)
        except ConnectionError as e:
            raise OSPDOpenvasError('Redis Error: Not possible '
                                   'to connect to the kb.') from e
        self.db_index = dbnum
        return ctx

    def db_find(self, patt):
        """ Search a pattern inside all kbs. When find it return it.
        """
        for i in range(0, self.max_dbindex):
            ctx = self.kb_connect(i)
            if ctx.keys(patt):
                return ctx

        return None

    def kb_new(self):
        """ Return a new kb context to an empty kb.
        """
        ctx = self.db_find(self.DBINDEX_NAME)
        for index in range(1, self.max_dbindex):
            if self.try_database_index(ctx, index):
                ctx = self.kb_connect(index)
                return ctx

        return None

    def get_kb_context(self):
        """ Get redis context if it is already connected or do a connection.
        """
        if self.rediscontext is not None:
            return self.rediscontext

        self.rediscontext = self.db_find(self.DBINDEX_NAME)

        if self.rediscontext is None:
            raise OSPDOpenvasError('Redis Error: Problem retrieving '
                                   'Redis Context')

        return self.rediscontext

    def select_kb(self, ctx, kbindex, set_global=False):
        """ Use an existent redis connection and select a redis kb.
        If needed, set the ctx as global.
        Arguments:
            ctx (redis obj): Redis context to use.
            kbindex (str): The new kb to select
            set_global (bool, optional): If should be the global context.
        """
        if not ctx:
            raise RequiredArgument('select_kb(): A valid Redis context is '
                                   'required.')
        if not kbindex:
            raise RequiredArgument('select_kb(): A valid KB index is '
                                   'required.')

        ctx.execute_command('SELECT ' + str(kbindex))
        if set_global:
            self.set_redisctx(ctx)
            self.db_index = str(kbindex)

    def get_list_item(self, name, ctx=None, start=LIST_FIRST_POS,
                      end=LIST_LAST_POS):
        """ Returns the specified elements from `start` to `end` of the
        list stored as `name`.

        Arguments:
            name (str): key name of a list.
            ctx (redis obj, optional): Redis context to use.
            start (int, optional): first range element to get.
            end (int, optional): last range element to get.

        Return List specified elements in the key.
        """
        if not name:
            raise RequiredArgument('get_list_item requires a name argument.')

        if not ctx:
            ctx = self.get_kb_context()
        return ctx.lrange(name, start, end)

    def remove_list_item(self, key, value, ctx=None):
        """ Remove item from the key list.
        Arguments:
            key (str): key name of a list.
            value (str): Value to be removed from the key.
            ctx (redis obj, optional): Redis context to use.
        """
        if not key:
            raise RequiredArgument('remove_list_item requires a key argument.')
        if not value:
            raise RequiredArgument('remove_list_item requires a value '
                                   'argument.')

        if not ctx:
            ctx = self.get_kb_context()
        ctx.lrem(key, count=LIST_ALL, value=value)

    def get_single_item(self, name, ctx=None, index=LIST_FIRST_POS):
        """ Get a single KB element.
        Arguments:
            name (str): key name of a list.
            ctx (redis obj, optional): Redis context to use.
            index (int, optional): index of the element to be return.
        Return an element.
        """
        if not name:
            raise RequiredArgument('get_single_item requires a name argument.')

        if not ctx:
            ctx = self.get_kb_context()
        return ctx.lindex(name, index)

    def add_single_item(self, name, values, ctx=None):
        """ Add a single KB element with one or more values.
        Arguments:
            name (str): key name of a list.
            value (list): Elements to add to the key.
            ctx (redis obj, optional): Redis context to use.
        """
        if not name:
            raise RequiredArgument('add_list_item requires a name argument.')
        if not values:
            raise RequiredArgument('add_list_item requires a value argument.')

        if not ctx:
            ctx = self.get_kb_context()
        ctx.rpush(name, *set(values))

    def set_single_item(self, name, value, ctx=None):
        """ Set (replace) a single KB element.
        Arguments:
            name (str): key name of a list.
            value (list): New elements to add to the key.
            ctx (redis obj, optional): Redis context to use.
        """
        if not name:
            raise RequiredArgument('set_single_item requires a name argument.')
        if not value:
            raise RequiredArgument('set_single_item requires a value argument.')

        if not ctx:
            ctx = self.get_kb_context()
        pipe = ctx.pipeline()
        pipe.delete(name)
        pipe.rpush(name, *set(value))
        pipe.execute()

    def get_pattern(self, pattern, ctx=None):
        """ Get all items stored under a given pattern.
        Arguments:
            pattern (str): key pattern to match.
            ctx (redis obj, optional): Redis context to use.
        Return a list with the elements under the matched key.
        """
        if not pattern:
            raise RequiredArgument('get_pattern requires a pattern argument.')

        if not ctx:
            ctx = self.get_kb_context()
        items = ctx.keys(pattern)

        elem_list = []
        for item in items:
            elem_list.append([
                item,
                ctx.lrange(item, start=LIST_FIRST_POS, end=LIST_LAST_POS),
            ])
        return elem_list

    def get_elem_pattern_by_index(self, pattern, index=1, ctx=None):
        """ Get all items with index 'index', stored under
        a given pattern.
        Arguments:
            pattern (str): key pattern to match.
            index (int, optional): Index of the element to get from the list.
            ctx (redis obj, optional): Redis context to use.
        Return a list with the elements under the matched key and given index.
        """
        if not pattern:
            raise RequiredArgument('get_elem_pattern_by_index '
                                   'requires a pattern argument.')

        if not ctx:
            ctx = self.get_kb_context()
        items = ctx.keys(pattern)

        elem_list = []
        for item in items:
            elem_list.append([item, ctx.lindex(item, index)])
        return elem_list

    def release_db(self, kbindex=0):
        """ Connect to redis and select the db by index.
        Flush db and delete the index from dbindex_name list.
        Arguments:
            kbindex (int, optional): KB index to flush and release.
        """
        ctx = self.kb_connect(kbindex)
        ctx.flushdb()
        ctx = self.kb_connect()
        ctx.hdel(self.DBINDEX_NAME, kbindex)

    def get_result(self, ctx=None):
        """ Get and remove the oldest result from the list.
        Arguments:
            ctx (redis obj, optional): Redis context to use.
        Return a list with scan results
        """
        if not ctx:
            ctx = self.get_kb_context()
        return ctx.rpop("internal/results")

    def get_status(self, ctx=None):
        """ Get and remove the oldest host scan status from the list.
        Arguments:
            ctx (redis obj, optional): Redis context to use.
        Return a string which represents the host scan status.
        """
        if not ctx:
            ctx = self.get_kb_context()
        return ctx.rpop("internal/status")

    def get_host_scan_scan_start_time(self, ctx=None):
        """ Get the timestamp of the scan start from redis.
        Arguments:
            ctx (redis obj, optional): Redis context to use.
        Return a string with the timestamp of the scan start.
        """
        if not ctx:
            ctx = self.get_kb_context()
        return ctx.rpop("internal/start_time")

    def get_host_scan_scan_end_time(self, ctx=None):
        """ Get the timestamp of the scan end from redis.
        Arguments:
            ctx (redis obj, optional): Redis context to use.
        Return a string with the timestamp of scan end .
        """
        if not ctx:
            ctx = self.get_kb_context()
        return ctx.rpop("internal/end_time")


1		# -- coding: utf-8 --
2		# Copyright (C) 2018 Greenbone Networks GmbH
3		#
4		# SPDX-License-Identifier: GPL-2.0-or-later
5		#
6		# This program is free software; you can redistribute it and/or
7		# modify it under the terms of the GNU General Public License
8		# as published by the Free Software Foundation; either version 2
9		# of the License, or (at your option) any later version.
10		#
11		# This program is distributed in the hope that it will be useful,
12		# but WITHOUT ANY WARRANTY; without even the implied warranty of
13		# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14		# GNU General Public License for more details.
15		#
16		# You should have received a copy of the GNU General Public License
17		# along with this program; if not, write to the Free Software
18		# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
19
20		""" Access management for redis-based OpenVAS Scanner Database."""
21
22		import redis
23		import subprocess
24
25		from ospd_openvas.errors import OSPDOpenvasError
26		from ospd_openvas.errors import RequiredArgument
27		from ospd.ospd import logger
28
29		SOCKET_TIMEOUT = 60 # in seconds
30		LIST_FIRST_POS = 0
31		LIST_LAST_POS = -1
32		LIST_ALL = 0
33
34		# Possible positions of nvt values in cache list.
35		NVT_META_FIELDS = [
36		"NVT_FILENAME_POS",
37		"NVT_REQUIRED_KEYS_POS",
38		"NVT_MANDATORY_KEYS_POS",
39		"NVT_EXCLUDED_KEYS_POS",
40		"NVT_REQUIRED_UDP_PORTS_POS",
41		"NVT_REQUIRED_PORTS_POS",
42		"NVT_DEPENDENCIES_POS",
43		"NVT_TAGS_POS",
44		"NVT_CVES_POS",
45		"NVT_BIDS_POS",
46		"NVT_XREFS_POS",
47		"NVT_CATEGORY_POS",
48		"NVT_TIMEOUT_POS",
49		"NVT_FAMILY_POS",
50		"NVT_NAME_POS",
51		]
52
53
54	View Code Duplication	class OpenvasDB(object):
		0 ignored issues – show Duplication introduced 2018-11-30 08:38 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
55		""" Class to connect to redis, to perform queries, and to move
56		from a KB to another."""
57		# Name of the namespace usage bitmap in redis.
58		DBINDEX_NAME = "GVM.__GlobalDBIndex"
59
60		def __init__(self):
61		# Path to the Redis socket.
62		self.db_address = None
63
64		self.max_dbindex = 0
65		self.db_index = 0
66		self.rediscontext = None
67
68		@staticmethod
69		def _parse_openvassd_db_address(result):
70		""" Return the path to the redis socket.
71		Arguments:
72		result (bytes) Output of `openvassd -s`
73		Return redis unix socket path.
74		"""
75		path = None
76		result = result.decode('ascii')
77		for conf in result.split('\n'):
78		if conf.find('db_address') == 0:
79		path = conf.split('=')
80		break
81
82		if not path:
83		raise OSPDOpenvasError('Redis Error: Not possible to '
84		'find the path to the redis socket.')
85		return path[1].strip()
86
87		def get_db_connection(self):
88		""" Retrieve the db address from openvassd config.
89		"""
90		result = subprocess.check_output(
91		['openvassd', '-s'], stderr=subprocess.STDOUT)
92
93		if result:
94		path = self._parse_openvassd_db_address(result)
95
96		self.db_address = path
		0 ignored issues – show introduced 2018-12-03 13:33 UTC by Report Bug Copy Issue Report The variable `path` does not seem to be defined in case `result` on line `93` is `False`. Are you sure this can never be the case? Loading history...
97
98		def max_db_index(self):
99		"""Set the number of databases have been configured into kbr struct.
100		"""
101		ctx = self.kb_connect()
102		resp = ctx.config_get('databases')
103
104		if len(resp) == 1:
105		self.max_dbindex = int(resp.get('databases'))
106		else:
107		raise OSPDOpenvasError('Redis Error: Not possible '
108		'to get max_dbindex.')
109
110		def set_redisctx(self, ctx):
111		""" Set the current rediscontext.
112		Arguments:
113		ctx (object): Redis context to be set as default.
114		"""
115		if not ctx:
116		raise RequiredArgument('set_redisctx: A valid Redis context is '
117		'required.')
118		self.rediscontext = ctx
119
120		def db_init(self):
121		""" Set db_address and max_db_index. """
122		self.get_db_connection()
123		self.max_db_index()
124
125		def try_database_index(self, ctx, kb):
126		""" Check if a redis kb is already in use. If not, set it
127		as in use and return.
128		Arguments:
129		ctx (object): Redis object connected to the kb with the
130		DBINDEX_NAME key.
131		kb (int): Kb number intended to be used.
132
133		Return True if it is possible to use the kb. False if the given kb
134		number is already in use.
135		"""
136		_IN_USE = 1
137		try:
138		resp = ctx.hsetnx(self.DBINDEX_NAME, kb, _IN_USE)
139		except:
140		raise OSPDOpenvasError('Redis Error: Not possible '
141		'to set %s.' % self.DBINDEX_NAME)
142
143		if resp == 1:
144		return True
145		return False
146
147		def kb_connect(self, dbnum=0):
148		""" Connect to redis to the given database or to the default db 0 .
149
150		Arguments:
151		dbnum (int, optional): The db number to connect to.
152
153		Return a redis context on success.
154		"""
155		self.get_db_connection()
156
157		try:
158		ctx = redis.Redis(unix_socket_path=self.db_address,
159		db=dbnum,
160		socket_timeout=SOCKET_TIMEOUT, charset="latin-1",
161		decode_responses=True)
162		except ConnectionError as e:
163		raise OSPDOpenvasError('Redis Error: Not possible '
164		'to connect to the kb.') from e
165		self.db_index = dbnum
166		return ctx
167
168		def db_find(self, patt):
169		""" Search a pattern inside all kbs. When find it return it.
170		"""
171		for i in range(0, self.max_dbindex):
172		ctx = self.kb_connect(i)
173		if ctx.keys(patt):
174		return ctx
175
176		return None
177
178		def kb_new(self):
179		""" Return a new kb context to an empty kb.
180		"""
181		ctx = self.db_find(self.DBINDEX_NAME)
182		for index in range(1, self.max_dbindex):
183		if self.try_database_index(ctx, index):
184		ctx = self.kb_connect(index)
185		return ctx
186
187		return None
188
189		def get_kb_context(self):
190		""" Get redis context if it is already connected or do a connection.
191		"""
192		if self.rediscontext is not None:
193		return self.rediscontext
194
195		self.rediscontext = self.db_find(self.DBINDEX_NAME)
196
197		if self.rediscontext is None:
198		raise OSPDOpenvasError('Redis Error: Problem retrieving '
199		'Redis Context')
200
201		return self.rediscontext
202
203		def select_kb(self, ctx, kbindex, set_global=False):
204		""" Use an existent redis connection and select a redis kb.
205		If needed, set the ctx as global.
206		Arguments:
207		ctx (redis obj): Redis context to use.
208		kbindex (str): The new kb to select
209		set_global (bool, optional): If should be the global context.
210		"""
211		if not ctx:
212		raise RequiredArgument('select_kb(): A valid Redis context is '
213		'required.')
214		if not kbindex:
215		raise RequiredArgument('select_kb(): A valid KB index is '
216		'required.')
217
218		ctx.execute_command('SELECT ' + str(kbindex))
219		if set_global:
220		self.set_redisctx(ctx)
221		self.db_index = str(kbindex)
222
223		def get_list_item(self, name, ctx=None, start=LIST_FIRST_POS,
224		end=LIST_LAST_POS):
225		""" Returns the specified elements from `start` to `end` of the
226		list stored as `name`.
227
228		Arguments:
229		name (str): key name of a list.
230		ctx (redis obj, optional): Redis context to use.
231		start (int, optional): first range element to get.
232		end (int, optional): last range element to get.
233
234		Return List specified elements in the key.
235		"""
236		if not name:
237		raise RequiredArgument('get_list_item requires a name argument.')
238
239		if not ctx:
240		ctx = self.get_kb_context()
241		return ctx.lrange(name, start, end)
242
243		def remove_list_item(self, key, value, ctx=None):
244		""" Remove item from the key list.
245		Arguments:
246		key (str): key name of a list.
247		value (str): Value to be removed from the key.
248		ctx (redis obj, optional): Redis context to use.
249		"""
250		if not key:
251		raise RequiredArgument('remove_list_item requires a key argument.')
252		if not value:
253		raise RequiredArgument('remove_list_item requires a value '
254		'argument.')
255
256		if not ctx:
257		ctx = self.get_kb_context()
258		ctx.lrem(key, count=LIST_ALL, value=value)
259
260		def get_single_item(self, name, ctx=None, index=LIST_FIRST_POS):
261		""" Get a single KB element.
262		Arguments:
263		name (str): key name of a list.
264		ctx (redis obj, optional): Redis context to use.
265		index (int, optional): index of the element to be return.
266		Return an element.
267		"""
268		if not name:
269		raise RequiredArgument('get_single_item requires a name argument.')
270
271		if not ctx:
272		ctx = self.get_kb_context()
273		return ctx.lindex(name, index)
274
275		def add_single_item(self, name, values, ctx=None):
276		""" Add a single KB element with one or more values.
277		Arguments:
278		name (str): key name of a list.
279		value (list): Elements to add to the key.
280		ctx (redis obj, optional): Redis context to use.
281		"""
282		if not name:
283		raise RequiredArgument('add_list_item requires a name argument.')
284		if not values:
285		raise RequiredArgument('add_list_item requires a value argument.')
286
287		if not ctx:
288		ctx = self.get_kb_context()
289		ctx.rpush(name, *set(values))
290
291		def set_single_item(self, name, value, ctx=None):
292		""" Set (replace) a single KB element.
293		Arguments:
294		name (str): key name of a list.
295		value (list): New elements to add to the key.
296		ctx (redis obj, optional): Redis context to use.
297		"""
298		if not name:
299		raise RequiredArgument('set_single_item requires a name argument.')
300		if not value:
301		raise RequiredArgument('set_single_item requires a value argument.')
302
303		if not ctx:
304		ctx = self.get_kb_context()
305		pipe = ctx.pipeline()
306		pipe.delete(name)
307		pipe.rpush(name, *set(value))
308		pipe.execute()
309
310		def get_pattern(self, pattern, ctx=None):
311		""" Get all items stored under a given pattern.
312		Arguments:
313		pattern (str): key pattern to match.
314		ctx (redis obj, optional): Redis context to use.
315		Return a list with the elements under the matched key.
316		"""
317		if not pattern:
318		raise RequiredArgument('get_pattern requires a pattern argument.')
319
320		if not ctx:
321		ctx = self.get_kb_context()
322		items = ctx.keys(pattern)
323
324		elem_list = []
325		for item in items:
326		elem_list.append([
327		item,
328		ctx.lrange(item, start=LIST_FIRST_POS, end=LIST_LAST_POS),
329		])
330		return elem_list
331
332		def get_elem_pattern_by_index(self, pattern, index=1, ctx=None):
333		""" Get all items with index 'index', stored under
334		a given pattern.
335		Arguments:
336		pattern (str): key pattern to match.
337		index (int, optional): Index of the element to get from the list.
338		ctx (redis obj, optional): Redis context to use.
339		Return a list with the elements under the matched key and given index.
340		"""
341		if not pattern:
342		raise RequiredArgument('get_elem_pattern_by_index '
343		'requires a pattern argument.')
344
345		if not ctx:
346		ctx = self.get_kb_context()
347		items = ctx.keys(pattern)
348
349		elem_list = []
350		for item in items:
351		elem_list.append([item, ctx.lindex(item, index)])
352		return elem_list
353
354		def release_db(self, kbindex=0):
355		""" Connect to redis and select the db by index.
356		Flush db and delete the index from dbindex_name list.
357		Arguments:
358		kbindex (int, optional): KB index to flush and release.
359		"""
360		ctx = self.kb_connect(kbindex)
361		ctx.flushdb()
362		ctx = self.kb_connect()
363		ctx.hdel(self.DBINDEX_NAME, kbindex)
364
365		def get_result(self, ctx=None):
366		""" Get and remove the oldest result from the list.
367		Arguments:
368		ctx (redis obj, optional): Redis context to use.
369		Return a list with scan results
370		"""
371		if not ctx:
372		ctx = self.get_kb_context()
373		return ctx.rpop("internal/results")
374
375		def get_status(self, ctx=None):
376		""" Get and remove the oldest host scan status from the list.
377		Arguments:
378		ctx (redis obj, optional): Redis context to use.
379		Return a string which represents the host scan status.
380		"""
381		if not ctx:
382		ctx = self.get_kb_context()
383		return ctx.rpop("internal/status")
384
385		def get_host_scan_scan_start_time(self, ctx=None):
386		""" Get the timestamp of the scan start from redis.
387		Arguments:
388		ctx (redis obj, optional): Redis context to use.
389		Return a string with the timestamp of the scan start.
390		"""
391		if not ctx:
392		ctx = self.get_kb_context()
393		return ctx.rpop("internal/start_time")
394
395		def get_host_scan_scan_end_time(self, ctx=None):
396		""" Get the timestamp of the scan end from redis.
397		Arguments:
398		ctx (redis obj, optional): Redis context to use.
399		Return a string with the timestamp of scan end .
400		"""
401		if not ctx:
402		ctx = self.get_kb_context()
403		return ctx.rpop("internal/end_time")
404

greenbone / ospd-openvas

Push — master ( b2d3e0...9e273c )

ospd_openvas.db.OpenvasDB.set_single_item() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like