ospd_openvas.db.ScanDB.get_host_ip() - Code Metrics - Inspection of "Merge pull request #285 from jjnicola/results-main..." - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 351cbc...3c6238 )

unknown

created 2020-07-15 15:20 UTC

ospd_openvas.db.ScanDB.get_host_ip() A

↳ Parent: ospd_openvas.db

Complexity

Conditions

Size

Total Lines	6
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	6
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) -> RedisCtx:
        """ Connect to redis to the given database or to the default db 0 .

        Arguments:
            dbnum: The db number to connect to.

        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="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)

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

greenbone / ospd-openvas

Push — master ( 351cbc...3c6238 )

ospd_openvas.db.ScanDB.get_host_ip() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like