ospd_openvas.db.OpenvasDB.get_host_ip() - Code Metrics - Inspection of "Merge pull request #64 from jjnicola/resume-tasks" - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 6c1e13...78bc3c )

by Juan José

created 2019-04-11 11:48 UTC

ospd_openvas.db.OpenvasDB.get_host_ip() A

↳ Parent: ospd_openvas.db

Complexity

Conditions

Size

Total Lines	9
Code Lines	4

Duplication

Lines	9
Ratio	100 %

Importance

Changes

Metric	Value
cc	2
eloc	4
nop	2
dl	9
loc	9
rs	10
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
import time

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_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()
        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:
                self.logger.debug('Redis connection lost. Trying again in 5 seconds.')
                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_list_item("internal/ip")


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

greenbone / ospd-openvas

Push — master ( 6c1e13...78bc3c )

ospd_openvas.db.OpenvasDB.get_host_ip() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like