ospd_openvas.db.KbDB.add_credentials_to_scan_preferences() - Code Metrics - Inspection of "Merge 20.08 into master" - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#307)

unknown

created 2020-08-04 12:08 UTC

KbDB.add_credentials_to_scan_preferences() A

↳ Parent: ospd_openvas.db

Complexity

Conditions

Size

Total Lines	9
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	3
dl	0
loc	9
rs	10
c	0
b	0
f	0

# -*- coding: utf-8 -*-
# Copyright (C) 2014-2020 Greenbone Networks GmbH
#
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.


""" Access management for redis-based OpenVAS Scanner Database."""
import logging
import sys
import time

from typing import List, NewType, Optional, Iterable, Iterator, Tuple

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

# Name of the namespace usage bitmap in redis.
DBINDEX_NAME = "GVM.__GlobalDBIndex"

logger = logging.getLogger(__name__)

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


class OpenvasDB:
    """ Class to connect to redis, to perform queries, and to move
    from a KB to another."""

    _db_address = None

    @classmethod
    def get_database_address(cls) -> Optional[str]:
        if not cls._db_address:
            settings = Openvas.get_settings()

            cls._db_address = settings.get('db_address')

        return cls._db_address

    @classmethod
    def create_context(
        cls, dbnum: Optional[int] = 0, encoding: Optional[str] = 'latin-1'
    ) -> RedisCtx:
        """ Connect to redis to the given database or to the default db 0 .

        Arguments:
            dbnum: The db number to connect to.
            encoding: The encoding to be used to read and write.

        Return a new redis context on success.
        """
        tries = 5
        while tries:
            try:
                ctx = redis.Redis(
                    unix_socket_path=cls.get_database_address(),
                    db=dbnum,
                    socket_timeout=SOCKET_TIMEOUT,
                    encoding=encoding,
                    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)

        return ctx


    @classmethod
    def find_database_by_pattern(
        cls, pattern: str, max_database_index: int
    ) -> Tuple[Optional[RedisCtx], Optional[int]]:
        """ Search a pattern inside all kbs up to max_database_index.

        Returns the redis context for the db and its index as a tuple or
        None, None if the db with the pattern couldn't be found.
        """
        for i in range(0, max_database_index):
            ctx = cls.create_context(i)
            if ctx.keys(pattern):
                return (ctx, i)

        return (None, None)

    @staticmethod
    def select_database(ctx: RedisCtx, kbindex: str):
        """ Use an existent redis connection and select a redis kb.

        Arguments:
            ctx: Redis context to use.
            kbindex: The new kb to select
        """
        if not ctx:
            raise RequiredArgument('select_database', 'ctx')
        if not kbindex:
            raise RequiredArgument('select_database', 'kbindex')

        ctx.execute_command('SELECT ' + str(kbindex))

    @staticmethod
    def get_list_item(
        ctx: RedisCtx,
        name: str,
        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:
            ctx: Redis context to use.
            name: key name of a list.
            start: first range element to get.
            end: last range element to get.

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

        return ctx.lrange(name, start, end)

    @staticmethod
    def get_last_list_item(ctx: RedisCtx, name: str) -> str:
        if not ctx:
            raise RequiredArgument('get_last_list_item', 'ctx')
        if not name:
            raise RequiredArgument('get_last_list_item', 'name')

        return ctx.rpop(name)

    @staticmethod
    def pop_list_items(ctx: RedisCtx, name: str) -> List[str]:
        if not ctx:
            raise RequiredArgument('pop_list_items', 'ctx')
        if not name:
            raise RequiredArgument('pop_list_items', 'name')

        pipe = ctx.pipeline()
        pipe.lrange(name, LIST_FIRST_POS, LIST_LAST_POS)
        pipe.delete(name)
        results, redis_return_code = pipe.execute()

        # The results are left-pushed. To preserver the order
        # the result list must be reversed.
        if redis_return_code:
            results.reverse()
        else:
            results = []

        return results

    @staticmethod
    def get_key_count(ctx: RedisCtx, pattern: Optional[str] = None) -> int:
        """ Get the number of keys matching with the pattern.

        Arguments:
            ctx: Redis context to use.
            pattern: pattern used as filter.
        """
        if not pattern:
            pattern = "*"

        if not ctx:
            raise RequiredArgument('get_key_count', 'ctx')

        return len(ctx.keys(pattern))

    @staticmethod
    def remove_list_item(ctx: RedisCtx, key: str, value: str):
        """ Remove item from the key list.

        Arguments:
            ctx: Redis context to use.
            key: key name of a list.
            value: Value to be removed from the key.
        """
        if not ctx:
            raise RequiredArgument('remove_list_item ', 'ctx')
        if not key:
            raise RequiredArgument('remove_list_item', 'key')
        if not value:
            raise RequiredArgument('remove_list_item ', 'value')

        ctx.lrem(key, count=LIST_ALL, value=value)

    @staticmethod
    def get_single_item(
        ctx: RedisCtx, name: str, index: Optional[int] = LIST_FIRST_POS,
    ) -> Optional[str]:
        """ Get a single KB element.

        Arguments:
            ctx: Redis context to use.
            name: key name of a list.
            index: index of the element to be return.
                   Defaults to the first element in the list.

        Return the first element of the list or None if the name couldn't be
        found.
        """
        if not ctx:
            raise RequiredArgument('get_single_item', 'ctx')
        if not name:
            raise RequiredArgument('get_single_item', 'name')

        return ctx.lindex(name, index)

    @staticmethod
    def add_single_item(ctx: RedisCtx, name: str, values: Iterable):
        """ Add a single KB element with one or more values.

        Arguments:
            ctx: Redis context to use.
            name: key name of a list.
            value: Elements to add to the key.
        """
        if not ctx:
            raise RequiredArgument('add_list_item', 'ctx')
        if not name:
            raise RequiredArgument('add_list_item', 'name')
        if not values:
            raise RequiredArgument('add_list_item', 'value')

        ctx.rpush(name, *set(values))

    @staticmethod
    def set_single_item(ctx: RedisCtx, name: str, value: Iterable):
        """ Set (replace) a single KB element.

        Arguments:
            ctx: Redis context to use.
            name: key name of a list.
            value: New elements to add to the key.
        """
        if not ctx:
            raise RequiredArgument('set_single_item', 'ctx')
        if not name:
            raise RequiredArgument('set_single_item', 'name')
        if not value:
            raise RequiredArgument('set_single_item', 'value')

        pipe = ctx.pipeline()
        pipe.delete(name)
        pipe.rpush(name, *set(value))
        pipe.execute()

    @staticmethod
    def get_pattern(ctx: RedisCtx, pattern: str) -> List:
        """ Get all items stored under a given pattern.

        Arguments:
            ctx: Redis context to use.
            pattern: key pattern to match.

        Return a list with the elements under the matched key.
        """
        if not ctx:
            raise RequiredArgument('get_pattern', 'ctx')
        if not pattern:
            raise RequiredArgument('get_pattern', 'pattern')

        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

    @classmethod
    def get_keys_by_pattern(cls, ctx: RedisCtx, pattern: str) -> List[str]:
        """ Get all items with index 'index', stored under
        a given pattern.

        Arguments:
            ctx: Redis context to use.
            pattern: key pattern to match.

        Return a sorted list with the elements under the matched key
        """
        if not ctx:
            raise RequiredArgument('get_elem_pattern_by_index', 'ctx')
        if not pattern:
            raise RequiredArgument('get_elem_pattern_by_index', 'pattern')

        return sorted(ctx.keys(pattern))

    @classmethod
    def get_filenames_and_oids(
        cls, ctx: RedisCtx,
    ) -> Iterable[Tuple[str, str]]:
        """ Get all items with index 'index', stored under
        a given pattern.

        Arguments:
            ctx: Redis context to use.

        Return an iterable where each single tuple contains the filename
            as first element and the oid as the second one.
        """
        if not ctx:
            raise RequiredArgument('get_filenames_and_oids', 'ctx')

        items = cls.get_keys_by_pattern(ctx, 'nvt:*')

        return ((ctx.lindex(item, 0), item[4:]) for item in items)



class BaseDB:
    def __init__(self, kbindex: int, ctx: Optional[RedisCtx] = None):
        if ctx is None:
            self.ctx = OpenvasDB.create_context(kbindex)
        else:
            self.ctx = ctx

        self.index = kbindex

    def flush(self):
        """ Flush the database """
        self.ctx.flushdb()


class BaseKbDB(BaseDB):
    def _add_single_item(
        self, name: str, values: Iterable, utf8_enc: Optional[bool] = False
    ):
        ''' Changing the encoding format of an existing redis context
        is not possible. Therefore a new temporary redis context is
        created to store key-values encoded with utf-8.'''
        if utf8_enc:
            ctx = OpenvasDB.create_context(self.index, encoding='utf-8')
            OpenvasDB.add_single_item(ctx, name, values)
        else:
            OpenvasDB.add_single_item(self.ctx, name, values)

    def _set_single_item(self, name: str, value: Iterable):
        """ Set (replace) a single KB element.

        Arguments:
            name: key name of a list.
            value: New elements to add to the key.
        """
        OpenvasDB.set_single_item(self.ctx, name, value)

    def _get_single_item(self, name: str) -> Optional[str]:
        """ Get a single KB element.

        Arguments:
            name: key name of a list.
        """
        return OpenvasDB.get_single_item(self.ctx, name)

    def _get_list_item(self, name: str,) -> Optional[List]:
        """ Returns the specified elements from `start` to `end` of the
        list stored as `name`.

        Arguments:
            name: key name of a list.

        Return List specified elements in the key.
        """
        return OpenvasDB.get_list_item(self.ctx, name)

    def _pop_list_items(self, name: str) -> List:
        return OpenvasDB.pop_list_items(self.ctx, name)

    def _remove_list_item(self, key: str, value: str):
        """ Remove item from the key list.

        Arguments:
            key: key name of a list.
            value: Value to be removed from the key.
        """
        OpenvasDB.remove_list_item(self.ctx, key, value)

    def get_result(self) -> Optional[str]:
        """ Get and remove the oldest result from the list.

        Return the oldest scan results
        """
        return self._pop_list_items("internal/results")

    def get_status(self, openvas_scan_id: str) -> Optional[str]:
        """ Return the status of the host scan """
        return self._get_single_item('internal/{}'.format(openvas_scan_id))

    def __repr__(self):
        return '<{} index={}>'.format(self.__class__.__name__, self.index)


class ScanDB(BaseKbDB):
    """ Database for a scanning a single host """

    def select(self, kbindex: int) -> "ScanDB":
        """ Select a redis kb.

        Arguments:
            kbindex: The new kb to select
        """
        OpenvasDB.select_database(self.ctx, kbindex)
        self.index = kbindex
        return self


class KbDB(BaseKbDB):
    def get_scan_databases(self) -> Iterator[ScanDB]:
        """ Returns an iterator yielding corresponding ScanDBs

        The returned Iterator can't be converted to an Iterable like a List.
        Each yielded ScanDB must be used independently in a for loop. If the
        Iterator gets converted into an Iterable all returned ScanDBs will use
        the same redis context pointing to the same redis database.
        """
        dbs = self._get_list_item('internal/dbindex')
        scan_db = ScanDB(self.index)
        for kbindex in dbs:
            if kbindex == self.index:
                continue

            yield scan_db.select(kbindex)

    def add_scan_id(self, scan_id: str, openvas_scan_id: str):
        self._add_single_item('internal/{}'.format(openvas_scan_id), ['new'])
        self._add_single_item(
            'internal/{}/globalscanid'.format(scan_id), [openvas_scan_id]
        )
        self._add_single_item('internal/scanid', [openvas_scan_id])

    def add_scan_preferences(self, openvas_scan_id: str, preferences: Iterable):
        self._add_single_item(
            'internal/{}/scanprefs'.format(openvas_scan_id), preferences
        )

    def add_credentials_to_scan_preferences(
        self, openvas_scan_id: str, preferences: Iterable
    ):
        ''' Force the usage of the utf-8 encoding, since some credentials
        contain special chars not supported by latin-1 encoding. '''
        self._add_single_item(
            'internal/{}/scanprefs'.format(openvas_scan_id),
            preferences,
            utf8_enc=True,
        )

    def add_scan_process_id(self, pid: int):
        self._add_single_item('internal/ovas_pid', [pid])

    def get_scan_process_id(self) -> Optional[str]:
        return self._get_single_item('internal/ovas_pid')

    def remove_scan_database(self, scan_db: ScanDB):
        self._remove_list_item('internal/dbindex', scan_db.index)

    def target_is_finished(self, scan_id: str) -> bool:
        """ Check if a target has finished. """

        openvas_scan_id = self._get_single_item(
            'internal/{}/globalscanid'.format(scan_id)
        )
        status = self._get_single_item('internal/{}'.format(openvas_scan_id))

        return status == 'finished' or status is None

    def stop_scan(self, openvas_scan_id: str):
        self._set_single_item(
            'internal/{}'.format(openvas_scan_id), ['stop_all']
        )

    def scan_is_stopped(self, openvas_scan_id: str) -> bool:
        """ Check if the scan should be stopped
        """
        status = self._get_single_item('internal/%s' % openvas_scan_id)
        return status == 'stop_all'

    def get_scan_status(self) -> List:
        """ Get and remove the oldest host scan status from the list.

        Return a string which represents the host scan status.
        """
        return self._pop_list_items("internal/status")


class MainDB(BaseDB):
    """ Main Database """

    DEFAULT_INDEX = 0

    def __init__(self, ctx=None):
        super().__init__(self.DEFAULT_INDEX, ctx)

        self._max_dbindex = None

    @property
    def max_database_index(self):
        """Set the number of databases have been configured into kbr struct.
        """
        if self._max_dbindex is None:
            resp = self.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.'
                )

        return self._max_dbindex

    def try_database(self, index: int) -> bool:
        """ Check if a redis db 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.
            index: Number intended to be used.

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

        return resp == 1

    def get_new_kb_database(self) -> Optional[KbDB]:
        """ Return a new kb db to an empty kb.
        """
        for index in range(1, self.max_database_index):
            if self.try_database(index):
                kbdb = KbDB(index)
                kbdb.flush()
                return kbdb

        return None

    def find_kb_database_by_scan_id(
        self, scan_id: str
    ) -> Tuple[Optional[str], Optional["KbDB"]]:
        """ Find a kb db by via a global scan id
        """
        for index in range(1, self.max_database_index):
            ctx = OpenvasDB.create_context(index)
            openvas_scan_id = OpenvasDB.get_single_item(
                ctx, 'internal/{}/globalscanid'.format(scan_id)
            )
            if openvas_scan_id:
                return (openvas_scan_id, KbDB(index, ctx))

        return (None, None)

    def release_database(self, database: BaseDB):
        self.release_database_by_index(database.index)
        database.flush()

    def release_database_by_index(self, index: int):
        self.ctx.hdel(DBINDEX_NAME, index)

    def release(self):
        self.release_database(self)


1			# -- coding: utf-8 --
2			# Copyright (C) 2014-2020 Greenbone Networks GmbH
3			#
4			# SPDX-License-Identifier: AGPL-3.0-or-later
5			#
6			# This program is free software: you can redistribute it and/or modify
7			# it under the terms of the GNU Affero General Public License as
8			# published by the Free Software Foundation, either version 3 of the
9			# 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 Affero General Public License for more details.
15			#
16			# You should have received a copy of the GNU Affero General Public License
17			# along with this program. If not, see <http://www.gnu.org/licenses/>.
18
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, Iterable, Iterator, Tuple
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			# Name of the namespace usage bitmap in redis.
58			DBINDEX_NAME = "GVM.__GlobalDBIndex"
59
60			logger = logging.getLogger(__name__)
61
62			# Types
63			RedisCtx = NewType('RedisCtx', redis.Redis)
64
65
66			class OpenvasDB:
67			""" Class to connect to redis, to perform queries, and to move
68			from a KB to another."""
69
70			_db_address = None
71
72			@classmethod
73			def get_database_address(cls) -> Optional[str]:
74			if not cls._db_address:
75			settings = Openvas.get_settings()
76
77			cls._db_address = settings.get('db_address')
78
79			return cls._db_address
80
81			@classmethod
82			def create_context(
83			cls, dbnum: Optional[int] = 0, encoding: Optional[str] = 'latin-1'
84			) -> RedisCtx:
85			""" Connect to redis to the given database or to the default db 0 .
86
87			Arguments:
88			dbnum: The db number to connect to.
89			encoding: The encoding to be used to read and write.
90
91			Return a new redis context on success.
92			"""
93			tries = 5
94			while tries:
95			try:
96			ctx = redis.Redis(
97			unix_socket_path=cls.get_database_address(),
98			db=dbnum,
99			socket_timeout=SOCKET_TIMEOUT,
100			encoding=encoding,
101			decode_responses=True,
102			)
103			ctx.keys("test")
104			except (redis.exceptions.ConnectionError, FileNotFoundError) as err:
105			logger.debug(
106			'Redis connection lost: %s. Trying again in 5 seconds.', err
107			)
108			tries = tries - 1
109			time.sleep(5)
110			continue
111			break
112
113			if not tries:
114			logger.error('Redis Error: Not possible to connect to the kb.')
115			sys.exit(1)
116
117			return ctx
			0 ignored issues – show introduced 2020-08-04 11:54 UTC by Report Bug Copy Issue Report The variable `ctx` does not seem to be defined in case the `while` loop on line `94` is not entered. Are you sure this can never be the case? Loading history...
118
119			@classmethod
120			def find_database_by_pattern(
121			cls, pattern: str, max_database_index: int
122			) -> Tuple[Optional[RedisCtx], Optional[int]]:
123			""" Search a pattern inside all kbs up to max_database_index.
124
125			Returns the redis context for the db and its index as a tuple or
126			None, None if the db with the pattern couldn't be found.
127			"""
128			for i in range(0, max_database_index):
129			ctx = cls.create_context(i)
130			if ctx.keys(pattern):
131			return (ctx, i)
132
133			return (None, None)
134
135			@staticmethod
136			def select_database(ctx: RedisCtx, kbindex: str):
137			""" Use an existent redis connection and select a redis kb.
138
139			Arguments:
140			ctx: Redis context to use.
141			kbindex: The new kb to select
142			"""
143			if not ctx:
144			raise RequiredArgument('select_database', 'ctx')
145			if not kbindex:
146			raise RequiredArgument('select_database', 'kbindex')
147
148			ctx.execute_command('SELECT ' + str(kbindex))
149
150			@staticmethod
151			def get_list_item(
152			ctx: RedisCtx,
153			name: str,
154			start: Optional[int] = LIST_FIRST_POS,
155			end: Optional[int] = LIST_LAST_POS,
156			) -> Optional[list]:
157			""" Returns the specified elements from `start` to `end` of the
158			list stored as `name`.
159
160			Arguments:
161			ctx: Redis context to use.
162			name: key name of a list.
163			start: first range element to get.
164			end: last range element to get.
165
166			Return List specified elements in the key.
167			"""
168			if not ctx:
169			raise RequiredArgument('get_list_item', 'ctx')
170			if not name:
171			raise RequiredArgument('get_list_item', 'name')
172
173			return ctx.lrange(name, start, end)
174
175			@staticmethod
176			def get_last_list_item(ctx: RedisCtx, name: str) -> str:
177			if not ctx:
178			raise RequiredArgument('get_last_list_item', 'ctx')
179			if not name:
180			raise RequiredArgument('get_last_list_item', 'name')
181
182			return ctx.rpop(name)
183
184			@staticmethod
185			def pop_list_items(ctx: RedisCtx, name: str) -> List[str]:
186			if not ctx:
187			raise RequiredArgument('pop_list_items', 'ctx')
188			if not name:
189			raise RequiredArgument('pop_list_items', 'name')
190
191			pipe = ctx.pipeline()
192			pipe.lrange(name, LIST_FIRST_POS, LIST_LAST_POS)
193			pipe.delete(name)
194			results, redis_return_code = pipe.execute()
195
196			# The results are left-pushed. To preserver the order
197			# the result list must be reversed.
198			if redis_return_code:
199			results.reverse()
200			else:
201			results = []
202
203			return results
204
205			@staticmethod
206			def get_key_count(ctx: RedisCtx, pattern: Optional[str] = None) -> int:
207			""" Get the number of keys matching with the pattern.
208
209			Arguments:
210			ctx: Redis context to use.
211			pattern: pattern used as filter.
212			"""
213			if not pattern:
214			pattern = "*"
215
216			if not ctx:
217			raise RequiredArgument('get_key_count', 'ctx')
218
219			return len(ctx.keys(pattern))
220
221			@staticmethod
222			def remove_list_item(ctx: RedisCtx, key: str, value: str):
223			""" Remove item from the key list.
224
225			Arguments:
226			ctx: Redis context to use.
227			key: key name of a list.
228			value: Value to be removed from the key.
229			"""
230			if not ctx:
231			raise RequiredArgument('remove_list_item ', 'ctx')
232			if not key:
233			raise RequiredArgument('remove_list_item', 'key')
234			if not value:
235			raise RequiredArgument('remove_list_item ', 'value')
236
237			ctx.lrem(key, count=LIST_ALL, value=value)
238
239			@staticmethod
240			def get_single_item(
241			ctx: RedisCtx, name: str, index: Optional[int] = LIST_FIRST_POS,
242			) -> Optional[str]:
243			""" Get a single KB element.
244
245			Arguments:
246			ctx: Redis context to use.
247			name: key name of a list.
248			index: index of the element to be return.
249			Defaults to the first element in the list.
250
251			Return the first element of the list or None if the name couldn't be
252			found.
253			"""
254			if not ctx:
255			raise RequiredArgument('get_single_item', 'ctx')
256			if not name:
257			raise RequiredArgument('get_single_item', 'name')
258
259			return ctx.lindex(name, index)
260
261			@staticmethod
262			def add_single_item(ctx: RedisCtx, name: str, values: Iterable):
263			""" Add a single KB element with one or more values.
264
265			Arguments:
266			ctx: Redis context to use.
267			name: key name of a list.
268			value: Elements to add to the key.
269			"""
270			if not ctx:
271			raise RequiredArgument('add_list_item', 'ctx')
272			if not name:
273			raise RequiredArgument('add_list_item', 'name')
274			if not values:
275			raise RequiredArgument('add_list_item', 'value')
276
277			ctx.rpush(name, *set(values))
278
279			@staticmethod
280			def set_single_item(ctx: RedisCtx, name: str, value: Iterable):
281			""" Set (replace) a single KB element.
282
283			Arguments:
284			ctx: Redis context to use.
285			name: key name of a list.
286			value: New elements to add to the key.
287			"""
288			if not ctx:
289			raise RequiredArgument('set_single_item', 'ctx')
290			if not name:
291			raise RequiredArgument('set_single_item', 'name')
292			if not value:
293			raise RequiredArgument('set_single_item', 'value')
294
295			pipe = ctx.pipeline()
296			pipe.delete(name)
297			pipe.rpush(name, *set(value))
298			pipe.execute()
299
300			@staticmethod
301			def get_pattern(ctx: RedisCtx, pattern: str) -> List:
302			""" Get all items stored under a given pattern.
303
304			Arguments:
305			ctx: Redis context to use.
306			pattern: key pattern to match.
307
308			Return a list with the elements under the matched key.
309			"""
310			if not ctx:
311			raise RequiredArgument('get_pattern', 'ctx')
312			if not pattern:
313			raise RequiredArgument('get_pattern', 'pattern')
314
315			items = ctx.keys(pattern)
316
317			elem_list = []
318			for item in items:
319			elem_list.append(
320			[
321			item,
322			ctx.lrange(item, start=LIST_FIRST_POS, end=LIST_LAST_POS),
323			]
324			)
325			return elem_list
326
327			@classmethod
328			def get_keys_by_pattern(cls, ctx: RedisCtx, pattern: str) -> List[str]:
329			""" Get all items with index 'index', stored under
330			a given pattern.
331
332			Arguments:
333			ctx: Redis context to use.
334			pattern: key pattern to match.
335
336			Return a sorted list with the elements under the matched key
337			"""
338			if not ctx:
339			raise RequiredArgument('get_elem_pattern_by_index', 'ctx')
340			if not pattern:
341			raise RequiredArgument('get_elem_pattern_by_index', 'pattern')
342
343			return sorted(ctx.keys(pattern))
344
345			@classmethod
346			def get_filenames_and_oids(
347			cls, ctx: RedisCtx,
348			) -> Iterable[Tuple[str, str]]:
349			""" Get all items with index 'index', stored under
350			a given pattern.
351
352			Arguments:
353			ctx: Redis context to use.
354
355			Return an iterable where each single tuple contains the filename
356			as first element and the oid as the second one.
357			"""
358			if not ctx:
359			raise RequiredArgument('get_filenames_and_oids', 'ctx')
360
361			items = cls.get_keys_by_pattern(ctx, 'nvt:*')
362
363			return ((ctx.lindex(item, 0), item[4:]) for item in items)
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-03-26 16:28 UTC by Report Bug Copy Issue Report The variable `item` does not seem to be defined. Loading history...
364
365
366			class BaseDB:
367			def __init__(self, kbindex: int, ctx: Optional[RedisCtx] = None):
368			if ctx is None:
369			self.ctx = OpenvasDB.create_context(kbindex)
370			else:
371			self.ctx = ctx
372
373			self.index = kbindex
374
375			def flush(self):
376			""" Flush the database """
377			self.ctx.flushdb()
378
379
380			class BaseKbDB(BaseDB):
381			def _add_single_item(
382			self, name: str, values: Iterable, utf8_enc: Optional[bool] = False
383			):
384			''' Changing the encoding format of an existing redis context
385			is not possible. Therefore a new temporary redis context is
386			created to store key-values encoded with utf-8.'''
387			if utf8_enc:
388			ctx = OpenvasDB.create_context(self.index, encoding='utf-8')
389			OpenvasDB.add_single_item(ctx, name, values)
390			else:
391			OpenvasDB.add_single_item(self.ctx, name, values)
392
393			def _set_single_item(self, name: str, value: Iterable):
394			""" Set (replace) a single KB element.
395
396			Arguments:
397			name: key name of a list.
398			value: New elements to add to the key.
399			"""
400			OpenvasDB.set_single_item(self.ctx, name, value)
401
402			def _get_single_item(self, name: str) -> Optional[str]:
403			""" Get a single KB element.
404
405			Arguments:
406			name: key name of a list.
407			"""
408			return OpenvasDB.get_single_item(self.ctx, name)
409
410			def _get_list_item(self, name: str,) -> Optional[List]:
411			""" Returns the specified elements from `start` to `end` of the
412			list stored as `name`.
413
414			Arguments:
415			name: key name of a list.
416
417			Return List specified elements in the key.
418			"""
419			return OpenvasDB.get_list_item(self.ctx, name)
420
421			def _pop_list_items(self, name: str) -> List:
422			return OpenvasDB.pop_list_items(self.ctx, name)
423
424			def _remove_list_item(self, key: str, value: str):
425			""" Remove item from the key list.
426
427			Arguments:
428			key: key name of a list.
429			value: Value to be removed from the key.
430			"""
431			OpenvasDB.remove_list_item(self.ctx, key, value)
432
433			def get_result(self) -> Optional[str]:
434			""" Get and remove the oldest result from the list.
435
436			Return the oldest scan results
437			"""
438			return self._pop_list_items("internal/results")
439
440			def get_status(self, openvas_scan_id: str) -> Optional[str]:
441			""" Return the status of the host scan """
442			return self._get_single_item('internal/{}'.format(openvas_scan_id))
443
444			def __repr__(self):
445			return '<{} index={}>'.format(self.__class__.__name__, self.index)
446
447
448			class ScanDB(BaseKbDB):
449			""" Database for a scanning a single host """
450
451			def select(self, kbindex: int) -> "ScanDB":
452			""" Select a redis kb.
453
454			Arguments:
455			kbindex: The new kb to select
456			"""
457			OpenvasDB.select_database(self.ctx, kbindex)
458			self.index = kbindex
459			return self
460
461
462			class KbDB(BaseKbDB):
463			def get_scan_databases(self) -> Iterator[ScanDB]:
464			""" Returns an iterator yielding corresponding ScanDBs
465
466			The returned Iterator can't be converted to an Iterable like a List.
467			Each yielded ScanDB must be used independently in a for loop. If the
468			Iterator gets converted into an Iterable all returned ScanDBs will use
469			the same redis context pointing to the same redis database.
470			"""
471			dbs = self._get_list_item('internal/dbindex')
472			scan_db = ScanDB(self.index)
473			for kbindex in dbs:
474			if kbindex == self.index:
475			continue
476
477			yield scan_db.select(kbindex)
478
479			def add_scan_id(self, scan_id: str, openvas_scan_id: str):
480			self._add_single_item('internal/{}'.format(openvas_scan_id), ['new'])
481			self._add_single_item(
482			'internal/{}/globalscanid'.format(scan_id), [openvas_scan_id]
483			)
484			self._add_single_item('internal/scanid', [openvas_scan_id])
485
486			def add_scan_preferences(self, openvas_scan_id: str, preferences: Iterable):
487			self._add_single_item(
488			'internal/{}/scanprefs'.format(openvas_scan_id), preferences
489			)
490
491			def add_credentials_to_scan_preferences(
492			self, openvas_scan_id: str, preferences: Iterable
493			):
494			''' Force the usage of the utf-8 encoding, since some credentials
495			contain special chars not supported by latin-1 encoding. '''
496			self._add_single_item(
497			'internal/{}/scanprefs'.format(openvas_scan_id),
498			preferences,
499			utf8_enc=True,
500			)
501
502			def add_scan_process_id(self, pid: int):
503			self._add_single_item('internal/ovas_pid', [pid])
504
505			def get_scan_process_id(self) -> Optional[str]:
506			return self._get_single_item('internal/ovas_pid')
507
508			def remove_scan_database(self, scan_db: ScanDB):
509			self._remove_list_item('internal/dbindex', scan_db.index)
510
511			def target_is_finished(self, scan_id: str) -> bool:
512			""" Check if a target has finished. """
513
514			openvas_scan_id = self._get_single_item(
515			'internal/{}/globalscanid'.format(scan_id)
516			)
517			status = self._get_single_item('internal/{}'.format(openvas_scan_id))
518
519			return status == 'finished' or status is None
520
521			def stop_scan(self, openvas_scan_id: str):
522			self._set_single_item(
523			'internal/{}'.format(openvas_scan_id), ['stop_all']
524			)
525
526			def scan_is_stopped(self, openvas_scan_id: str) -> bool:
527			""" Check if the scan should be stopped
528			"""
529			status = self._get_single_item('internal/%s' % openvas_scan_id)
530			return status == 'stop_all'
531
532			def get_scan_status(self) -> List:
533			""" Get and remove the oldest host scan status from the list.
534
535			Return a string which represents the host scan status.
536			"""
537			return self._pop_list_items("internal/status")
538
539
540			class MainDB(BaseDB):
541			""" Main Database """
542
543			DEFAULT_INDEX = 0
544
545			def __init__(self, ctx=None):
546			super().__init__(self.DEFAULT_INDEX, ctx)
547
548			self._max_dbindex = None
549
550			@property
551			def max_database_index(self):
552			"""Set the number of databases have been configured into kbr struct.
553			"""
554			if self._max_dbindex is None:
555			resp = self.ctx.config_get('databases')
556
557			if len(resp) == 1:
558			self._max_dbindex = int(resp.get('databases'))
559			else:
560			raise OspdOpenvasError(
561			'Redis Error: Not possible to get max_dbindex.'
562			)
563
564			return self._max_dbindex
565
566			def try_database(self, index: int) -> bool:
567			""" Check if a redis db is already in use. If not, set it
568			as in use and return.
569
570			Arguments:
571			ctx: Redis object connected to the kb with the
572			DBINDEX_NAME key.
573			index: Number intended to be used.
574
575			Return True if it is possible to use the db. False if the given db
576			number is already in use.
577			"""
578			_in_use = 1
579			try:
580			resp = self.ctx.hsetnx(DBINDEX_NAME, index, _in_use)
581			except:
582			raise OspdOpenvasError(
583			'Redis Error: Not possible to set %s.' % DBINDEX_NAME
584			)
585
586			return resp == 1
587
588			def get_new_kb_database(self) -> Optional[KbDB]:
589			""" Return a new kb db to an empty kb.
590			"""
591			for index in range(1, self.max_database_index):
592			if self.try_database(index):
593			kbdb = KbDB(index)
594			kbdb.flush()
595			return kbdb
596
597			return None
598
599			def find_kb_database_by_scan_id(
600			self, scan_id: str
601			) -> Tuple[Optional[str], Optional["KbDB"]]:
602			""" Find a kb db by via a global scan id
603			"""
604			for index in range(1, self.max_database_index):
605			ctx = OpenvasDB.create_context(index)
606			openvas_scan_id = OpenvasDB.get_single_item(
607			ctx, 'internal/{}/globalscanid'.format(scan_id)
608			)
609			if openvas_scan_id:
610			return (openvas_scan_id, KbDB(index, ctx))
611
612			return (None, None)
613
614			def release_database(self, database: BaseDB):
615			self.release_database_by_index(database.index)
616			database.flush()
617
618			def release_database_by_index(self, index: int):
619			self.ctx.hdel(DBINDEX_NAME, index)
620
621			def release(self):
622			self.release_database(self)
623

greenbone / ospd-openvas

Pull Request — master (#307)

KbDB.add_credentials_to_scan_preferences() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like