ospd_openvas.db.OpenvasDB.get_key_count() - Code Metrics - Inspection of "Improve nvt feed check." - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#195)

by Juan José

created 2020-02-17 07:38 UTC

ospd_openvas.db.OpenvasDB.get_key_count() A

↳ Parent: ospd_openvas.db

Complexity

Conditions

Size

Total Lines	16
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	7
nop	3
dl	0
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 logging
import sys
import time

from typing import List, NewType, Optional

import redis

from ospd.errors import RequiredArgument
from ospd_openvas.errors import OspdOpenvasError
from ospd_openvas.openvas import Openvas

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",
]

logger = logging.getLogger(__name__)

# Types
RedisCtx = NewType('RedisCtx', redis.Redis)


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

    def get_db_connection(self):
        """ Retrieve the db address from openvas config.
        """
        if self.db_address:
            return

        settings = Openvas.get_settings()

        if settings:
            self.db_address = settings.get('db_address')

    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: RedisCtx):
        """ Set the current rediscontext.
        Arguments:
            ctx: Redis context to be set as default.
        """
        if not ctx:
            raise RequiredArgument('set_redisctx', 'ctx')
        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: RedisCtx, kb: int) -> bool:
        """ Check if a redis kb is already in use. If not, set it
        as in use and return.
        Arguments:
            ctx: Redis object connected to the kb with the
                DBINDEX_NAME key.
            kb: 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: Optional[int] = 0) -> RedisCtx:
        """ Connect to redis to the given database or to the default db 0 .

        Arguments:
            dbnum: 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:
                logger.debug(
                    'Redis connection lost: %s. Trying again in 5 seconds.', err
                )
                tries = tries - 1
                time.sleep(5)
                continue
            break

        if not tries:
            logger.error('Redis Error: Not possible to connect to the kb.')
            sys.exit(1)

        self.db_index = dbnum
        return ctx


    def db_find(self, patt: str) -> Optional[RedisCtx]:
        """ 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) -> Optional[RedisCtx]:
        """ 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) -> RedisCtx:
        """ 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: RedisCtx, kbindex: str, set_global: bool = False):
        """ Use an existent redis connection and select a redis kb.
        If needed, set the ctx as global.
        Arguments:
            ctx: Redis context to use.
            kbindex: The new kb to select
            set_global: If should be the global context.
        """
        if not ctx:
            raise RequiredArgument('select_kb', 'ctx')
        if not kbindex:
            raise RequiredArgument('select_kb', 'kbindex')

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

    def get_list_item(
        self,
        name: str,
        ctx: Optional[RedisCtx] = None,
        start: Optional[int] = LIST_FIRST_POS,
        end: Optional[int] = LIST_LAST_POS,
    ) -> Optional[list]:
        """ Returns the specified elements from `start` to `end` of the
        list stored as `name`.

        Arguments:
            name: key name of a list.
            ctx: Redis context to use.
            start: first range element to get.
            end: last range element to get.

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

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

    def get_key_count(
        self, pattern: Optional[str] = None, ctx: Optional[RedisCtx] = None,
    ) -> Optional[int]:
        """ Get the number of keys matching with the pattern.
        Arguments:
            pattern: pattern used as filter.
            ctx: Redis context to use.
        Return an element.
        """
        if not pattern:
            pattern = "*"

        if not ctx:
            ctx = self.get_kb_context()

        return len(ctx.keys(pattern))

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

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

    def get_single_item(
        self,
        name: str,
        ctx: Optional[RedisCtx] = None,
        index: Optional[int] = LIST_FIRST_POS,
    ) -> Optional[str]:
        """ Get a single KB element.
        Arguments:
            name: key name of a list.
            ctx: Redis context to use.
            index: index of the element to be return.
        Return an element.
        """
        if not name:
            raise RequiredArgument('get_single_item', 'name')

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

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

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

    def set_single_item(
        self, name: str, value: List, ctx: Optional[RedisCtx] = None
    ):
        """ Set (replace) a single KB element.
        Arguments:
            name: key name of a list.
            value: New elements to add to the key.
            ctx: Redis context to use.
        """
        if not name:
            raise RequiredArgument('set_single_item', 'name')
        if not value:
            raise RequiredArgument('set_single_item', 'value')

        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: str, ctx: Optional[RedisCtx] = None) -> List:
        """ Get all items stored under a given pattern.
        Arguments:
            pattern: key pattern to match.
            ctx: Redis context to use.
        Return a list with the elements under the matched key.
        """
        if not pattern:
            raise RequiredArgument('get_pattern', 'pattern')

        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: str,
        index: Optional[int] = 1,
        ctx: Optional[RedisCtx] = None,
    ) -> List:
        """ Get all items with index 'index', stored under
        a given pattern.
        Arguments:
            pattern: key pattern to match.
            index: Index of the element to get from the list.
            ctx: 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', 'pattern')

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

greenbone / ospd-openvas

Pull Request — master (#195)

ospd_openvas.db.OpenvasDB.get_key_count() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like