ospd_openvas.openvas.Openvas.get_gvm_libs_version() - Code Metrics - Inspection of "Merge pull request #318 from jjnicola/cache-versio..." - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 7a7f2e...ca40a9 )

by Juan José

created 2020-08-17 08:47 UTC

ospd_openvas.openvas.Openvas.get_gvm_libs_version() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	16
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	11
nop	1
dl	0
loc	16
rs	9.85
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/>.


import logging
import subprocess

from typing import Optional, Dict, Any

logger = logging.getLogger(__name__)

_BOOL_DICT = {'no': 0, 'yes': 1}


class Openvas:
    """ Class for calling the openvas executable
    """

    @staticmethod
    def _get_version_output() -> Optional[str]:
        try:
            result = subprocess.check_output(
                ['openvas', '-V'], stderr=subprocess.STDOUT
            )
            return result.decode('ascii')
        except (subprocess.SubprocessError, OSError) as e:
            logger.debug(
                'Is was not possible to call openvas to get the version '
                'information. Reason %s',
                e,
            )
            return None

    @staticmethod
    def check() -> bool:
        """ Checks that openvas command line tool is found and
        is executable.
        """
        try:
            subprocess.check_call(['openvas', '-V'], stdout=subprocess.DEVNULL)
            return True
        except (subprocess.SubprocessError, OSError) as e:
            logger.debug(
                'It was not possible to call the openvas executable. Reason %s',
                e,
            )
            return False

    @staticmethod
    def check_sudo() -> bool:
        """ Checks if openvas can be run with sudo
        """
        try:
            subprocess.check_call(
                ['sudo', '-n', 'openvas', '-s'], stdout=subprocess.DEVNULL
            )
            return True
        except (subprocess.SubprocessError, OSError) as e:
            logger.debug(
                'It was not possible to call openvas with sudo. '
                'The scanner will run as non-root user. Reason %s',
                e,
            )
            return False

    @classmethod
    def get_version(cls) -> Optional[str]:
        """ Returns the version string of the openvas executable
        """
        result = cls._get_version_output()

        if result is None:
            return None

        version = result.split('\n')
        if version[0].find('OpenVAS') < 0:
            return None

        return version[0]

    @staticmethod
    def get_settings() -> Dict[str, Any]:
        """ Parses the current settings of the openvas executable
        """
        param_list = dict()

        try:
            result = subprocess.check_output(['openvas', '-s'])
            result = result.decode('ascii')
        except (subprocess.SubprocessError, OSError) as e:
            logger.warning('Could not gather openvas settings. Reason %s', e)
            return param_list

        for conf in result.split('\n'):
            if not conf:
                continue

            try:
                key, value = conf.split('=', 1)
            except ValueError:
                logger.warning("Could not parse openvas setting '%s'", conf)
                continue

            key = key.strip()
            value = value.strip()

            if value:
                value = _BOOL_DICT.get(value, value)
                param_list[key] = value

        return param_list

    @staticmethod
    def load_vts_into_redis():
        """ Loads all VTs into the redis database
        """
        logger.debug('Loading VTs into Redis DB...')

        try:
            subprocess.check_call(
                ['openvas', '--update-vt-info'], stdout=subprocess.DEVNULL
            )
            logger.debug('Finished loading VTs into Redis DB')
        except (subprocess.SubprocessError, OSError) as err:
            logger.error('OpenVAS Scanner failed to load VTs. %s', err)

    @staticmethod
    def start_scan(
        scan_id: str, sudo: bool = False, niceness: int = None
    ) -> Optional[subprocess.Popen]:
        """ Calls openvas to start a scan process
        """
        cmd = []

        if niceness:
            cmd += ['nice', '-n', niceness]
            logger.debug("Starting scan with niceness %s", niceness)

        if sudo:
            cmd += ['sudo', '-n']

        cmd += ['openvas', '--scan-start', scan_id]

        try:
            return subprocess.Popen(cmd, shell=False)
        except (subprocess.SubprocessError, OSError) as e:
            # the command is not available
            logger.warning("Could not start scan process. Reason %s", e)
            return None

    @staticmethod
    def stop_scan(scan_id: str, sudo: bool = False) -> bool:
        """ Calls openvas to stop a scan process
        """
        cmd = []

        if sudo:
            cmd += ['sudo', '-n']

        cmd += ['openvas', '--scan-stop', scan_id]

        try:
            subprocess.check_call(cmd)
            return True
        except (subprocess.SubprocessError, OSError) as e:
            # the command is not available
            logger.warning(
                'Not possible to stop scan: %s. Reason %s', scan_id, e,
            )
            return False


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			import logging
21			import subprocess
22
23			from typing import Optional, Dict, Any
24
25			logger = logging.getLogger(__name__)
26
27			_BOOL_DICT = {'no': 0, 'yes': 1}
28
29
30			class Openvas:
31			""" Class for calling the openvas executable
32			"""
33
34			@staticmethod
35			def _get_version_output() -> Optional[str]:
36			try:
37			result = subprocess.check_output(
38			['openvas', '-V'], stderr=subprocess.STDOUT
39			)
40			return result.decode('ascii')
41			except (subprocess.SubprocessError, OSError) as e:
42			logger.debug(
43			'Is was not possible to call openvas to get the version '
44			'information. Reason %s',
45			e,
46			)
47			return None
48
49			@staticmethod
50			def check() -> bool:
51			""" Checks that openvas command line tool is found and
52			is executable.
53			"""
54			try:
55			subprocess.check_call(['openvas', '-V'], stdout=subprocess.DEVNULL)
56			return True
57			except (subprocess.SubprocessError, OSError) as e:
58			logger.debug(
59			'It was not possible to call the openvas executable. Reason %s',
60			e,
61			)
62			return False
63
64			@staticmethod
65			def check_sudo() -> bool:
66			""" Checks if openvas can be run with sudo
67			"""
68			try:
69			subprocess.check_call(
70			['sudo', '-n', 'openvas', '-s'], stdout=subprocess.DEVNULL
71			)
72			return True
73			except (subprocess.SubprocessError, OSError) as e:
74			logger.debug(
75			'It was not possible to call openvas with sudo. '
76			'The scanner will run as non-root user. Reason %s',
77			e,
78			)
79			return False
80
81			@classmethod
82			def get_version(cls) -> Optional[str]:
83			""" Returns the version string of the openvas executable
84			"""
85			result = cls._get_version_output()
86
87			if result is None:
88			return None
89
90			version = result.split('\n')
91			if version[0].find('OpenVAS') < 0:
92			return None
93
94			return version[0]
95
96			@staticmethod
97			def get_settings() -> Dict[str, Any]:
98			""" Parses the current settings of the openvas executable
99			"""
100			param_list = dict()
101
102			try:
103			result = subprocess.check_output(['openvas', '-s'])
104			result = result.decode('ascii')
105			except (subprocess.SubprocessError, OSError) as e:
106			logger.warning('Could not gather openvas settings. Reason %s', e)
107			return param_list
108
109			for conf in result.split('\n'):
110			if not conf:
111			continue
112
113			try:
114			key, value = conf.split('=', 1)
115			except ValueError:
116			logger.warning("Could not parse openvas setting '%s'", conf)
117			continue
118
119			key = key.strip()
120			value = value.strip()
121
122			if value:
123			value = _BOOL_DICT.get(value, value)
124			param_list[key] = value
125
126			return param_list
127
128			@staticmethod
129			def load_vts_into_redis():
130			""" Loads all VTs into the redis database
131			"""
132			logger.debug('Loading VTs into Redis DB...')
133
134			try:
135			subprocess.check_call(
136			['openvas', '--update-vt-info'], stdout=subprocess.DEVNULL
137			)
138			logger.debug('Finished loading VTs into Redis DB')
139			except (subprocess.SubprocessError, OSError) as err:
140			logger.error('OpenVAS Scanner failed to load VTs. %s', err)
141
142			@staticmethod
143			def start_scan(
144			scan_id: str, sudo: bool = False, niceness: int = None
145			) -> Optional[subprocess.Popen]:
146			""" Calls openvas to start a scan process
147			"""
148			cmd = []
149
150			if niceness:
151			cmd += ['nice', '-n', niceness]
152			logger.debug("Starting scan with niceness %s", niceness)
153
154			if sudo:
155			cmd += ['sudo', '-n']
156
157			cmd += ['openvas', '--scan-start', scan_id]
158
159			try:
160			return subprocess.Popen(cmd, shell=False)
161			except (subprocess.SubprocessError, OSError) as e:
162			# the command is not available
163			logger.warning("Could not start scan process. Reason %s", e)
164			return None
165
166			@staticmethod
167			def stop_scan(scan_id: str, sudo: bool = False) -> bool:
168			""" Calls openvas to stop a scan process
169			"""
170			cmd = []
171
172			if sudo:
173			cmd += ['sudo', '-n']
174
175			cmd += ['openvas', '--scan-stop', scan_id]
176
177			try:
178			subprocess.check_call(cmd)
179			return True
180			except (subprocess.SubprocessError, OSError) as e:
181			# the command is not available
182			logger.warning(
183			'Not possible to stop scan: %s. Reason %s', scan_id, e,
184			)
185			return False
186

greenbone / ospd-openvas

Push — master ( 7a7f2e...ca40a9 )

ospd_openvas.openvas.Openvas.get_gvm_libs_version() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like