gvm.protocols.ospv1.Osp.start_scan() - Code Metrics - Inspection of "Connection shutdown" - greenbone/python-gvm - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#55)

unknown

created 2018-11-28 11:57 UTC

gvm.protocols.ospv1.Osp.start_scan() C

↳ Parent: gvm.protocols.ospv1

Complexity

Conditions

Size

Total Lines	97
Code Lines	33

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
eloc	33
nop	8
dl	0
loc	97
rs	6.6666
c	0
b	0
f	0

How to fix Long Method Many Parameters

# -*- coding: utf-8 -*-
# Copyright (C) 2018 Greenbone Networks GmbH
#
# SPDX-License-Identifier: GPL-3.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 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
"""
Module for communication to a daemon speaking `Open Scanner Protocol version 1`_

.. _Open Scanner Protocol version 1:
    https://docs.greenbone.net/API/GMP/gmp-7.0.html
"""
import logging

from gvm.errors import RequiredArgument
from gvm.utils import get_version_string
from gvm.xml import XmlCommand

from .base import GvmProtocol

logger = logging.getLogger(__name__)

PROTOCOL_VERSION = (1, 2,)


def create_credentials_element(_xmlcredentials, credentials):
    """Generates an xml element with credentials."""
    for service, credential in credentials.items():
        cred_type = credential.get('type')
        serv_port = credential.get('port')
        username = credential.get('username')
        password = credential.get('password')
        _xmlcredential = _xmlcredentials.add_element(
            'credential', attrs={
                'type': cred_type,
                'port': serv_port,
                'service': service,
            })
        _xmlcredential.add_element('username', username)
        _xmlcredential.add_element('password', password)
    return _xmlcredentials

def create_vt_selection_element(_xmlvtselection, vt_selection):
    """Generates an xml element with a selection of Vulnerability tests."""
    for vt_id, vt_values in vt_selection.items():
        if vt_id != 'vt_groups':
            _xmlvt = _xmlvtselection.add_element('vt_single',
                                                 attrs={'id': vt_id})
            if vt_values:
                for key, value in vt_values.items():
                    _xmlvt.add_element('vt_value', value, attrs={'id': key})
        else:
            for group in vt_values:
                _xmlvt = _xmlvtselection.add_element(
                    'vt_group', attrs={'filter': group})

    return _xmlvtselection

class Osp(GvmProtocol):

    @staticmethod
    def get_protocol_version():
        """Allow to determine the Open Scanner Protocol version.

            Returns:
                str: Implemented version of the Open Scanner Protocol
        """
        return get_version_string(PROTOCOL_VERSION)

    def _read(self):
        # OSP is stateless. Therefore the connection is closed after each
        # response and we must reset the connection
        data = super()._read()
        self.disconnect()
        return data

    def _send(self, data):
        # OSP is stateless. Therefore we can shutdown the socket if we are done
        # with sending
        super()._send(data)
        self._connection.finish_send()

    def get_version(self):
        """Get the version of the OSPD server which is connected to."""
        cmd = XmlCommand('get_version')
        return self._send_xml_command(cmd)

    def help(self):
        """Get the help text."""
        cmd = XmlCommand('help')
        return self._send_xml_command(cmd)

    def get_scans(self, scan_id=None, details=True, pop_results=False):
        """Get the stored scans.

        Arguments:
            scan_id (str, optional): UUID identifier for a scan.
            details (boolean, optional): Whether to get full scan reports.
                Default: True
            pop_results (boolean, optional) Whether to remove the fetched
                results. Default: False

        Returns:
            str: Response from server.
        """
        cmd = XmlCommand('get_scans')
        if scan_id:
            cmd.set_attribute('scan_id', scan_id)
        if details:
            cmd.set_attribute('details', '1')
        else:
            cmd.set_attribute('details', '0')

        if pop_results:
            cmd.set_attribute('pop_results', '1')
        else:
            cmd.set_attribute('pop_results', '0')

        return self._send_xml_command(cmd)

    def delete_scan(self, scan_id=None):
        """Delete a finished scan.

        Arguments:
            scan_id (str): UUID identifier for a finished scan.

        Returns:
            str: Response from server.
        """
        if not scan_id:
            raise ValueError('delete_scan requires a scan_id element')
        cmd = XmlCommand('delete_scan')
        cmd.set_attribute('scan_id', scan_id)

        return self._send_xml_command(cmd)

    def get_scanner_details(self):
        """Return scanner description and parameters."""
        cmd = XmlCommand('get_scanner_details')
        return self._send_xml_command(cmd)

    def get_vts(self, vt_id=None):
        """Return information about vulnerability tests,
        if offered by scanner.

        Arguments:
            vt_id (str, optional): UUID identifier for a vulnerability test.

        Returns:
            str: Response from server.
        """
        cmd = XmlCommand('get_vts')
        if vt_id:
            cmd.set_attribute('vt_id', vt_id)

        return self._send_xml_command(cmd)

    def start_scan(self, scan_id=None, parallel=1, target=None,

                   ports=None, targets=None, scanner_params=None,
                   vt_selection=None):
        """Start a new scan.

        Arguments:
            scan_id (str, optional): UUID identifier for a running scan.
            parallel (int, optional): Number of parallel scanned targets.
                Default 1.
            target (dict, optional): Deprecated. Please use targets instead.
            targets (list, optional): List of dictionaries. See example.
            ports (str, optional): Deprecated. Ports to use for target
                parameter.
            scanner_params: (dict, optional): Dictionary of scanner parameters.
            vt_selection: (dict, optional): Vulnerability tests to select. See
                example.

        Returns:
            str: Response from server.


        Examples:

            Scanner Parameters::

                scanner_parameters = {
                    'scan_param1': 'scan_param1_value',
                    'scan_param2': 'scan_param2_value',
                }

            Targets::

                targets = [{
                    'hosts': 'localhost',
                    'ports': '80,43'
                }, {
                    'hosts': '192.168.0.0/24',
                    'ports': '22',
                }, {
                    'credentials': {
                        'smb': {
                            'password': 'pass',
                            'port': 'port',
                            'type': 'type',
                            'username': 'username',
                        }
                    }
                }]

            VT Selection::

                vt_selection = {
                    'vt1': {},
                    'vt2': {'value_id': 'value'},
                    'vt_groups': ['family=debian', 'family=general']
                }
        """
        cmd = XmlCommand('start_scan')

        if scan_id:
            cmd.set_attribute('scan_id', scan_id)

        cmd.set_attribute('parallel', str(parallel))

        # Add <scanner_params> even if it is empty, since it is mandatory
        _xmlscanparams = cmd.add_element('scanner_params')
        if scanner_params:
            _xmlscanparams.set_attributes(scanner_params)

        if targets:
            _xmltargets = cmd.add_element('targets')
            for target in targets:

                _xmltarget = _xmltargets.add_element('target')
                hosts = target.get('hosts')
                ports = target.get('ports')
                credentials = target.get('credentials')
                _xmltarget.add_element('hosts', hosts)
                _xmltarget.add_element('ports', ports)
                if credentials:
                    _xmlcredentials = _xmltarget.add_element('credentials')
                    _xmlcredentials = (create_credentials_element(
                        _xmlcredentials, credentials))
        # Check target as attribute for legacy mode compatibility. Deprecated.
        elif target:
            cmd.set_attribute('target', target)
            if ports:
                cmd.set_attribute('ports', ports)
        else:
            raise RequiredArgument('start_scan requires a target. Please pass '
                                   'targets parameter.')

        if vt_selection:
            _xmlvtselection = cmd.add_element('vt_selection')
            _xmlvtselection = create_vt_selection_element(
                _xmlvtselection, vt_selection)

        return self._send_xml_command(cmd)

    def stop_scan(self, scan_id):
        """Stop a currently running scan.

        Args:
            scan_id (str): UUID identifier for a running scan.

        Returns:
            str: Response from server.
        """
        if not scan_id:
            raise RequiredArgument('stop_scan requires a scan_id argument')

        cmd = XmlCommand('stop_scan')
        cmd.set_attribute('scan_id', scan_id)

        return self._send_xml_command(cmd)


1			# -- coding: utf-8 --
2			# Copyright (C) 2018 Greenbone Networks GmbH
3			#
4			# SPDX-License-Identifier: GPL-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 General Public License as published by
8			# the Free Software Foundation, either version 3 of the License, or
9			# (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, see <http://www.gnu.org/licenses/>.
18			"""
19			Module for communication to a daemon speaking `Open Scanner Protocol version 1`_
20
21			.. _Open Scanner Protocol version 1:
22			https://docs.greenbone.net/API/GMP/gmp-7.0.html
23			"""
24			import logging
25
26			from gvm.errors import RequiredArgument
27			from gvm.utils import get_version_string
28			from gvm.xml import XmlCommand
29
30			from .base import GvmProtocol
31
32			logger = logging.getLogger(__name__)
33
34			PROTOCOL_VERSION = (1, 2,)
35
36
37			def create_credentials_element(_xmlcredentials, credentials):
38			"""Generates an xml element with credentials."""
39			for service, credential in credentials.items():
40			cred_type = credential.get('type')
41			serv_port = credential.get('port')
42			username = credential.get('username')
43			password = credential.get('password')
44			_xmlcredential = _xmlcredentials.add_element(
45			'credential', attrs={
46			'type': cred_type,
47			'port': serv_port,
48			'service': service,
49			})
50			_xmlcredential.add_element('username', username)
51			_xmlcredential.add_element('password', password)
52			return _xmlcredentials
53
54			def create_vt_selection_element(_xmlvtselection, vt_selection):
55			"""Generates an xml element with a selection of Vulnerability tests."""
56			for vt_id, vt_values in vt_selection.items():
57			if vt_id != 'vt_groups':
58			_xmlvt = _xmlvtselection.add_element('vt_single',
59			attrs={'id': vt_id})
60			if vt_values:
61			for key, value in vt_values.items():
62			_xmlvt.add_element('vt_value', value, attrs={'id': key})
63			else:
64			for group in vt_values:
65			_xmlvt = _xmlvtselection.add_element(
66			'vt_group', attrs={'filter': group})
67
68			return _xmlvtselection
69
70			class Osp(GvmProtocol):
71
72			@staticmethod
73			def get_protocol_version():
74			"""Allow to determine the Open Scanner Protocol version.
75
76			Returns:
77			str: Implemented version of the Open Scanner Protocol
78			"""
79			return get_version_string(PROTOCOL_VERSION)
80
81			def _read(self):
82			# OSP is stateless. Therefore the connection is closed after each
83			# response and we must reset the connection
84			data = super()._read()
85			self.disconnect()
86			return data
87
88			def _send(self, data):
89			# OSP is stateless. Therefore we can shutdown the socket if we are done
90			# with sending
91			super()._send(data)
92			self._connection.finish_send()
93
94			def get_version(self):
95			"""Get the version of the OSPD server which is connected to."""
96			cmd = XmlCommand('get_version')
97			return self._send_xml_command(cmd)
98
99			def help(self):
100			"""Get the help text."""
101			cmd = XmlCommand('help')
102			return self._send_xml_command(cmd)
103
104			def get_scans(self, scan_id=None, details=True, pop_results=False):
105			"""Get the stored scans.
106
107			Arguments:
108			scan_id (str, optional): UUID identifier for a scan.
109			details (boolean, optional): Whether to get full scan reports.
110			Default: True
111			pop_results (boolean, optional) Whether to remove the fetched
112			results. Default: False
113
114			Returns:
115			str: Response from server.
116			"""
117			cmd = XmlCommand('get_scans')
118			if scan_id:
119			cmd.set_attribute('scan_id', scan_id)
120			if details:
121			cmd.set_attribute('details', '1')
122			else:
123			cmd.set_attribute('details', '0')
124
125			if pop_results:
126			cmd.set_attribute('pop_results', '1')
127			else:
128			cmd.set_attribute('pop_results', '0')
129
130			return self._send_xml_command(cmd)
131
132			def delete_scan(self, scan_id=None):
133			"""Delete a finished scan.
134
135			Arguments:
136			scan_id (str): UUID identifier for a finished scan.
137
138			Returns:
139			str: Response from server.
140			"""
141			if not scan_id:
142			raise ValueError('delete_scan requires a scan_id element')
143			cmd = XmlCommand('delete_scan')
144			cmd.set_attribute('scan_id', scan_id)
145
146			return self._send_xml_command(cmd)
147
148			def get_scanner_details(self):
149			"""Return scanner description and parameters."""
150			cmd = XmlCommand('get_scanner_details')
151			return self._send_xml_command(cmd)
152
153			def get_vts(self, vt_id=None):
154			"""Return information about vulnerability tests,
155			if offered by scanner.
156
157			Arguments:
158			vt_id (str, optional): UUID identifier for a vulnerability test.
159
160			Returns:
161			str: Response from server.
162			"""
163			cmd = XmlCommand('get_vts')
164			if vt_id:
165			cmd.set_attribute('vt_id', vt_id)
166
167			return self._send_xml_command(cmd)
168
169			def start_scan(self, scan_id=None, parallel=1, target=None,
			0 ignored issues – show Comprehensibility introduced 2018-11-02 15:34 UTC by Report Bug Copy Issue Report This function exceeds the maximum number of variables (17/15). Loading history...
170			ports=None, targets=None, scanner_params=None,
171			vt_selection=None):
172			"""Start a new scan.
173
174			Arguments:
175			scan_id (str, optional): UUID identifier for a running scan.
176			parallel (int, optional): Number of parallel scanned targets.
177			Default 1.
178			target (dict, optional): Deprecated. Please use targets instead.
179			targets (list, optional): List of dictionaries. See example.
180			ports (str, optional): Deprecated. Ports to use for target
181			parameter.
182			scanner_params: (dict, optional): Dictionary of scanner parameters.
183			vt_selection: (dict, optional): Vulnerability tests to select. See
184			example.
185
186			Returns:
187			str: Response from server.
188
189
190			Examples:
191
192			Scanner Parameters::
193
194			scanner_parameters = {
195			'scan_param1': 'scan_param1_value',
196			'scan_param2': 'scan_param2_value',
197			}
198
199			Targets::
200
201			targets = [{
202			'hosts': 'localhost',
203			'ports': '80,43'
204			}, {
205			'hosts': '192.168.0.0/24',
206			'ports': '22',
207			}, {
208			'credentials': {
209			'smb': {
210			'password': 'pass',
211			'port': 'port',
212			'type': 'type',
213			'username': 'username',
214			}
215			}
216			}]
217
218			VT Selection::
219
220			vt_selection = {
221			'vt1': {},
222			'vt2': {'value_id': 'value'},
223			'vt_groups': ['family=debian', 'family=general']
224			}
225			"""
226			cmd = XmlCommand('start_scan')
227
228			if scan_id:
229			cmd.set_attribute('scan_id', scan_id)
230
231			cmd.set_attribute('parallel', str(parallel))
232
233			# Add <scanner_params> even if it is empty, since it is mandatory
234			_xmlscanparams = cmd.add_element('scanner_params')
235			if scanner_params:
236			_xmlscanparams.set_attributes(scanner_params)
237
238			if targets:
239			_xmltargets = cmd.add_element('targets')
240			for target in targets:
			0 ignored issues – show unused-code introduced 2018-11-02 15:34 UTC by Report Bug Copy Issue Report Redefining argument with the local name 'target' Loading history...
241			_xmltarget = _xmltargets.add_element('target')
242			hosts = target.get('hosts')
243			ports = target.get('ports')
244			credentials = target.get('credentials')
245			_xmltarget.add_element('hosts', hosts)
246			_xmltarget.add_element('ports', ports)
247			if credentials:
248			_xmlcredentials = _xmltarget.add_element('credentials')
249			_xmlcredentials = (create_credentials_element(
250			_xmlcredentials, credentials))
251			# Check target as attribute for legacy mode compatibility. Deprecated.
252			elif target:
253			cmd.set_attribute('target', target)
254			if ports:
255			cmd.set_attribute('ports', ports)
256			else:
257			raise RequiredArgument('start_scan requires a target. Please pass '
258			'targets parameter.')
259
260			if vt_selection:
261			_xmlvtselection = cmd.add_element('vt_selection')
262			_xmlvtselection = create_vt_selection_element(
263			_xmlvtselection, vt_selection)
264
265			return self._send_xml_command(cmd)
266
267			def stop_scan(self, scan_id):
268			"""Stop a currently running scan.
269
270			Args:
271			scan_id (str): UUID identifier for a running scan.
272
273			Returns:
274			str: Response from server.
275			"""
276			if not scan_id:
277			raise RequiredArgument('stop_scan requires a scan_id argument')
278
279			cmd = XmlCommand('stop_scan')
280			cmd.set_attribute('scan_id', scan_id)
281
282			return self._send_xml_command(cmd)
283

greenbone / python-gvm

Pull Request — master (#55)

gvm.protocols.ospv1.Osp.start_scan() C

Complexity

Size

Duplication

Importance

How to fix Long Method Many Parameters

Long Method

Many Parameters

Duplication Side-by-Side

Filter issues like