ospd_openvas.db.OpenvasDB.remove_list_item() - Code Metrics - Inspection of "Merge pull request #79 from janowagner/new_openvas" - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 0e3b51...d60795 )

by Juan José

created 2019-06-07 06:35 UTC

ospd_openvas.db.OpenvasDB.remove_list_item() A

↳ Parent: ospd_openvas.db

Complexity

Conditions

Size

Total Lines	16
Code Lines	8

Duplication

Lines	16
Ratio	100 %

Importance

Changes

Metric	Value
cc	4
eloc	8
nop	4
dl	16
loc	16
rs	10
c	0
b	0
f	0

# -*- coding: utf-8 -*-
# Copyright (C) 2018-2019 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
import time
import sys

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, wrapper_logger=None):
        # Path to the Redis socket.
        self.db_address = None

        self.max_dbindex = 0
        self.db_index = 0
        self.rediscontext = None
        self.logger = wrapper_logger.getChild('db') if wrapper_logger else logger

    @staticmethod
    def _parse_openvas_db_address(result):
        """ Return the path to the redis socket.
        Arguments:
            result (bytes) Output of `openvas -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 openvas config.
        """
        try:
            result = subprocess.check_output(
                ['openvas', '-s'], stderr=subprocess.STDOUT)
        except PermissionError:
            sys.exit("ERROR: %s: Not possible to run openvas. "
                     "Check permissions and/or path to the binary." \
                     % self.get_db_connection.__name__)
        if result:
            path = self._parse_openvas_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()
        tries = 5
        while (tries):
            try:
                ctx = redis.Redis(unix_socket_path=self.db_address,
                                  db=dbnum,
                                  socket_timeout=SOCKET_TIMEOUT, encoding="latin-1",
                                  decode_responses=True)
                ctx.keys("test")
            except (redis.exceptions.ConnectionError, FileNotFoundError) as err:
                self.logger.debug('Redis connection lost: %s. Trying again in 5 seconds.', err)
                tries = tries - 1
                time.sleep(5)
                continue
            break

        if not tries:
            raise OSPDOpenvasError('Redis Error: Not possible '
                                   'to connect to the kb.')


        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)
                ctx.flushdb()
                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")

    def get_host_ip(self, ctx=None):
        """ Get the ip of host_kb.
        Arguments:
            ctx (redis obj, optional): Redis context to use.
        Return a string with the ip of the host being scanned.
        """
        if not ctx:
            ctx = self.get_kb_context()
        return self.get_single_item("internal/ip")


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

greenbone / ospd-openvas

Push — master ( 0e3b51...d60795 )

ospd_openvas.db.OpenvasDB.remove_list_item() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like