ospd_openvas.db.OpenvasDB.get_filenames_and_oids() - Code Metrics - Inspection of "Merge pull request #225 from jjnicola/sorted-oid" - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 2fc1a9...66c124 )

by Juan José

created 2020-03-27 07:53 UTC

ospd_openvas.db.OpenvasDB.get_filenames_and_oids() A

↳ Parent: ospd_openvas.db

Complexity

Conditions

Size

Total Lines	19
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	7
nop	2
dl	0
loc	19
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, 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 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 _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 OpenvasDB.get_last_list_item(self.ctx, "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

    def get_scan_id(self):
        return self._get_single_item('internal/scan_id')

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

        Return a string which represents the host scan status.
        """
        return OpenvasDB.get_last_list_item(self.ctx, "internal/status")

    def get_host_ip(self) -> Optional[str]:
        """ Get the ip of host_kb.

        Return a string with the ip of the host being scanned.
        """
        return self._get_single_item("internal/ip")

    def get_host_scan_start_time(self) -> Optional[str]:
        """ Get the timestamp of the scan start from redis.

        Return a string with the timestamp of the scan start.
        """
        return OpenvasDB.get_last_list_item(self.ctx, "internal/start_time")

    def get_host_scan_end_time(self) -> Optional[str]:
        """ Get the timestamp of the scan end from redis.

        Return a string with the timestamp of scan end .
        """
        return OpenvasDB.get_last_list_item(self.ctx, "internal/end_time")

    def host_is_finished(self, openvas_scan_id: str) -> bool:
        """ Returns true if the scan of the host is finished """
        status = self.get_status(openvas_scan_id)
        return status == 'finished'


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'


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) 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, 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 get_key_count(ctx: RedisCtx, pattern: Optional[str] = None) -> int:
183			""" Get the number of keys matching with the pattern.
184
185			Arguments:
186			ctx: Redis context to use.
187			pattern: pattern used as filter.
188			"""
189			if not pattern:
190			pattern = "*"
191
192			if not ctx:
193			raise RequiredArgument('get_key_count', 'ctx')
194
195			return len(ctx.keys(pattern))
196
197			@staticmethod
198			def remove_list_item(ctx: RedisCtx, key: str, value: str):
199			""" Remove item from the key list.
200
201			Arguments:
202			ctx: Redis context to use.
203			key: key name of a list.
204			value: Value to be removed from the key.
205			"""
206			if not ctx:
207			raise RequiredArgument('remove_list_item ', 'ctx')
208			if not key:
209			raise RequiredArgument('remove_list_item', 'key')
210			if not value:
211			raise RequiredArgument('remove_list_item ', 'value')
212
213			ctx.lrem(key, count=LIST_ALL, value=value)
214
215			@staticmethod
216			def get_single_item(
217			ctx: RedisCtx, name: str, index: Optional[int] = LIST_FIRST_POS,
218			) -> Optional[str]:
219			""" Get a single KB element.
220
221			Arguments:
222			ctx: Redis context to use.
223			name: key name of a list.
224			index: index of the element to be return.
225			Defaults to the first element in the list.
226
227			Return the first element of the list or None if the name couldn't be
228			found.
229			"""
230			if not ctx:
231			raise RequiredArgument('get_single_item', 'ctx')
232			if not name:
233			raise RequiredArgument('get_single_item', 'name')
234
235			return ctx.lindex(name, index)
236
237			@staticmethod
238			def add_single_item(ctx: RedisCtx, name: str, values: Iterable):
239			""" Add a single KB element with one or more values.
240
241			Arguments:
242			ctx: Redis context to use.
243			name: key name of a list.
244			value: Elements to add to the key.
245			"""
246			if not ctx:
247			raise RequiredArgument('add_list_item', 'ctx')
248			if not name:
249			raise RequiredArgument('add_list_item', 'name')
250			if not values:
251			raise RequiredArgument('add_list_item', 'value')
252
253			ctx.rpush(name, *set(values))
254
255			@staticmethod
256			def set_single_item(ctx: RedisCtx, name: str, value: Iterable):
257			""" Set (replace) a single KB element.
258
259			Arguments:
260			ctx: Redis context to use.
261			name: key name of a list.
262			value: New elements to add to the key.
263			"""
264			if not ctx:
265			raise RequiredArgument('set_single_item', 'ctx')
266			if not name:
267			raise RequiredArgument('set_single_item', 'name')
268			if not value:
269			raise RequiredArgument('set_single_item', 'value')
270
271			pipe = ctx.pipeline()
272			pipe.delete(name)
273			pipe.rpush(name, *set(value))
274			pipe.execute()
275
276			@staticmethod
277			def get_pattern(ctx: RedisCtx, pattern: str) -> List:
278			""" Get all items stored under a given pattern.
279
280			Arguments:
281			ctx: Redis context to use.
282			pattern: key pattern to match.
283
284			Return a list with the elements under the matched key.
285			"""
286			if not ctx:
287			raise RequiredArgument('get_pattern', 'ctx')
288			if not pattern:
289			raise RequiredArgument('get_pattern', 'pattern')
290
291			items = ctx.keys(pattern)
292
293			elem_list = []
294			for item in items:
295			elem_list.append(
296			[
297			item,
298			ctx.lrange(item, start=LIST_FIRST_POS, end=LIST_LAST_POS),
299			]
300			)
301			return elem_list
302
303			@classmethod
304			def get_keys_by_pattern(cls, ctx: RedisCtx, pattern: str) -> List[str]:
305			""" Get all items with index 'index', stored under
306			a given pattern.
307
308			Arguments:
309			ctx: Redis context to use.
310			pattern: key pattern to match.
311
312			Return a sorted list with the elements under the matched key
313			"""
314			if not ctx:
315			raise RequiredArgument('get_elem_pattern_by_index', 'ctx')
316			if not pattern:
317			raise RequiredArgument('get_elem_pattern_by_index', 'pattern')
318
319			return sorted(ctx.keys(pattern))
320
321			@classmethod
322			def get_filenames_and_oids(
323			cls, ctx: RedisCtx,
324			) -> Iterable[Tuple[str, str]]:
325			""" Get all items with index 'index', stored under
326			a given pattern.
327
328			Arguments:
329			ctx: Redis context to use.
330
331			Return an iterable where each single tuple contains the filename
332			as first element and the oid as the second one.
333			"""
334			if not ctx:
335			raise RequiredArgument('get_filenames_and_oids', 'ctx')
336
337			items = cls.get_keys_by_pattern(ctx, 'nvt:*')
338
339			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...
340
341
342			class BaseDB:
343			def __init__(self, kbindex: int, ctx: Optional[RedisCtx] = None):
344			if ctx is None:
345			self.ctx = OpenvasDB.create_context(kbindex)
346			else:
347			self.ctx = ctx
348
349			self.index = kbindex
350
351			def flush(self):
352			""" Flush the database """
353			self.ctx.flushdb()
354
355
356			class BaseKbDB(BaseDB):
357			def _add_single_item(self, name: str, values: Iterable):
358			OpenvasDB.add_single_item(self.ctx, name, values)
359
360			def _set_single_item(self, name: str, value: Iterable):
361			""" Set (replace) a single KB element.
362
363			Arguments:
364			name: key name of a list.
365			value: New elements to add to the key.
366			"""
367			OpenvasDB.set_single_item(self.ctx, name, value)
368
369			def _get_single_item(self, name: str) -> Optional[str]:
370			""" Get a single KB element.
371
372			Arguments:
373			name: key name of a list.
374			"""
375			return OpenvasDB.get_single_item(self.ctx, name)
376
377			def _get_list_item(self, name: str,) -> Optional[List]:
378			""" Returns the specified elements from `start` to `end` of the
379			list stored as `name`.
380
381			Arguments:
382			name: key name of a list.
383
384			Return List specified elements in the key.
385			"""
386			return OpenvasDB.get_list_item(self.ctx, name)
387
388			def _remove_list_item(self, key: str, value: str):
389			""" Remove item from the key list.
390
391			Arguments:
392			key: key name of a list.
393			value: Value to be removed from the key.
394			"""
395			OpenvasDB.remove_list_item(self.ctx, key, value)
396
397			def get_result(self) -> Optional[str]:
398			""" Get and remove the oldest result from the list.
399
400			Return the oldest scan results
401			"""
402			return OpenvasDB.get_last_list_item(self.ctx, "internal/results")
403
404			def get_status(self, openvas_scan_id: str) -> Optional[str]:
405			""" Return the status of the host scan """
406			return self._get_single_item('internal/{}'.format(openvas_scan_id))
407
408			def __repr__(self):
409			return '<{} index={}>'.format(self.__class__.__name__, self.index)
410
411
412			class ScanDB(BaseKbDB):
413			""" Database for a scanning a single host """
414
415			def select(self, kbindex: int) -> "ScanDB":
416			""" Select a redis kb.
417
418			Arguments:
419			kbindex: The new kb to select
420			"""
421			OpenvasDB.select_database(self.ctx, kbindex)
422			self.index = kbindex
423			return self
424
425			def get_scan_id(self):
426			return self._get_single_item('internal/scan_id')
427
428			def get_scan_status(self) -> Optional[str]:
429			""" Get and remove the oldest host scan status from the list.
430
431			Return a string which represents the host scan status.
432			"""
433			return OpenvasDB.get_last_list_item(self.ctx, "internal/status")
434
435			def get_host_ip(self) -> Optional[str]:
436			""" Get the ip of host_kb.
437
438			Return a string with the ip of the host being scanned.
439			"""
440			return self._get_single_item("internal/ip")
441
442			def get_host_scan_start_time(self) -> Optional[str]:
443			""" Get the timestamp of the scan start from redis.
444
445			Return a string with the timestamp of the scan start.
446			"""
447			return OpenvasDB.get_last_list_item(self.ctx, "internal/start_time")
448
449			def get_host_scan_end_time(self) -> Optional[str]:
450			""" Get the timestamp of the scan end from redis.
451
452			Return a string with the timestamp of scan end .
453			"""
454			return OpenvasDB.get_last_list_item(self.ctx, "internal/end_time")
455
456			def host_is_finished(self, openvas_scan_id: str) -> bool:
457			""" Returns true if the scan of the host is finished """
458			status = self.get_status(openvas_scan_id)
459			return status == 'finished'
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_scan_process_id(self, pid: int):
492			self._add_single_item('internal/ovas_pid', [pid])
493
494			def get_scan_process_id(self) -> Optional[str]:
495			return self._get_single_item('internal/ovas_pid')
496
497			def remove_scan_database(self, scan_db: ScanDB):
498			self._remove_list_item('internal/dbindex', scan_db.index)
499
500			def target_is_finished(self, scan_id: str) -> bool:
501			""" Check if a target has finished. """
502
503			openvas_scan_id = self._get_single_item(
504			'internal/{}/globalscanid'.format(scan_id)
505			)
506			status = self._get_single_item('internal/{}'.format(openvas_scan_id))
507
508			return status == 'finished' or status is None
509
510			def stop_scan(self, openvas_scan_id: str):
511			self._set_single_item(
512			'internal/{}'.format(openvas_scan_id), ['stop_all']
513			)
514
515			def scan_is_stopped(self, openvas_scan_id: str) -> bool:
516			""" Check if the scan should be stopped
517			"""
518			status = self._get_single_item('internal/%s' % openvas_scan_id)
519			return status == 'stop_all'
520
521
522			class MainDB(BaseDB):
523			""" Main Database """
524
525			DEFAULT_INDEX = 0
526
527			def __init__(self, ctx=None):
528			super().__init__(self.DEFAULT_INDEX, ctx)
529
530			self._max_dbindex = None
531
532			@property
533			def max_database_index(self):
534			"""Set the number of databases have been configured into kbr struct.
535			"""
536			if self._max_dbindex is None:
537			resp = self.ctx.config_get('databases')
538
539			if len(resp) == 1:
540			self._max_dbindex = int(resp.get('databases'))
541			else:
542			raise OspdOpenvasError(
543			'Redis Error: Not possible to get max_dbindex.'
544			)
545
546			return self._max_dbindex
547
548			def try_database(self, index: int) -> bool:
549			""" Check if a redis db is already in use. If not, set it
550			as in use and return.
551
552			Arguments:
553			ctx: Redis object connected to the kb with the
554			DBINDEX_NAME key.
555			index: Number intended to be used.
556
557			Return True if it is possible to use the db. False if the given db
558			number is already in use.
559			"""
560			_in_use = 1
561			try:
562			resp = self.ctx.hsetnx(DBINDEX_NAME, index, _in_use)
563			except:
564			raise OspdOpenvasError(
565			'Redis Error: Not possible to set %s.' % DBINDEX_NAME
566			)
567
568			return resp == 1
569
570			def get_new_kb_database(self) -> Optional[KbDB]:
571			""" Return a new kb db to an empty kb.
572			"""
573			for index in range(1, self.max_database_index):
574			if self.try_database(index):
575			kbdb = KbDB(index)
576			kbdb.flush()
577			return kbdb
578
579			return None
580
581			def find_kb_database_by_scan_id(
582			self, scan_id: str
583			) -> Tuple[Optional[str], Optional["KbDB"]]:
584			""" Find a kb db by via a global scan id
585			"""
586			for index in range(1, self.max_database_index):
587			ctx = OpenvasDB.create_context(index)
588			openvas_scan_id = OpenvasDB.get_single_item(
589			ctx, 'internal/{}/globalscanid'.format(scan_id)
590			)
591			if openvas_scan_id:
592			return (openvas_scan_id, KbDB(index, ctx))
593
594			return (None, None)
595
596			def release_database(self, database: BaseDB):
597			self.release_database_by_index(database.index)
598			database.flush()
599
600			def release_database_by_index(self, index: int):
601			self.ctx.hdel(DBINDEX_NAME, index)
602
603			def release(self):
604			self.release_database(self)
605

greenbone / ospd-openvas

Push — master ( 2fc1a9...66c124 )

ospd_openvas.db.OpenvasDB.get_filenames_and_oids() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like