ospd.protocol.OspRequest.process_target_element() - Code Metrics - Inspection of "Update start scan parallel" - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#213)

unknown

created 2020-02-25 15:30 UTC

ospd.protocol.OspRequest.process_target_element() D

↳ Parent: ospd.protocol

Complexity

Conditions

Size

Total Lines	92
Code Lines	35

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	12
eloc	35
nop	2
dl	0
loc	92
rs	4.8
c	0
b	0
f	0

How to fix Long Method Complexity

# Copyright (C) 2020 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.

""" Helper classes for parsing and creating OSP XML requests and responses
"""

from typing import Dict, Union, List, Any

from xml.etree.ElementTree import SubElement, Element

from ospd.errors import OspdError


class OspRequest:
    @staticmethod
    def process_vts_params(
        scanner_vts: Element,
    ) -> Dict[str, Union[Dict, List]]:
        """ Receive an XML object with the Vulnerability Tests an their
        parameters to be use in a scan and return a dictionary.

        @param: XML element with vt subelements. Each vt has an
                id attribute. Optional parameters can be included
                as vt child.
                Example form:
                <vt_selection>
                  <vt_single id='vt1' />
                  <vt_single id='vt2'>
                    <vt_value id='param1'>value</vt_value>
                  </vt_single>
                  <vt_group filter='family=debian'/>
                  <vt_group filter='family=general'/>
                </vt_selection>

        @return: Dictionary containing the vts attribute and subelements,
                 like the VT's id and VT's parameters.
                 Example form:
                 {'vt1': {},
                  'vt2': {'value_id': 'value'},
                  'vt_groups': ['family=debian', 'family=general']}
        """
        vt_selection = {}  # type: Dict
        filters = []

        for vt in scanner_vts:
            if vt.tag == 'vt_single':
                vt_id = vt.attrib.get('id')
                vt_selection[vt_id] = {}

                for vt_value in vt:
                    if not vt_value.attrib.get('id'):
                        raise OspdError(
                            'Invalid VT preference. No attribute id'
                        )

                    vt_value_id = vt_value.attrib.get('id')
                    vt_value_value = vt_value.text if vt_value.text else ''
                    vt_selection[vt_id][vt_value_id] = vt_value_value

            if vt.tag == 'vt_group':
                vts_filter = vt.attrib.get('filter', None)

                if vts_filter is None:
                    raise OspdError('Invalid VT group. No filter given.')

                filters.append(vts_filter)

        vt_selection['vt_groups'] = filters

        return vt_selection

    @staticmethod
    def process_credentials_elements(cred_tree: Element) -> Dict:
        """ Receive an XML object with the credentials to run
        a scan against a given target.

        @param:
        <credentials>
          <credential type="up" service="ssh" port="22">
            <username>scanuser</username>
            <password>mypass</password>
          </credential>
          <credential type="up" service="smb">
            <username>smbuser</username>
            <password>mypass</password>
          </credential>
        </credentials>

        @return: Dictionary containing the credentials for a given target.
                 Example form:
                 {'ssh': {'type': type,
                          'port': port,
                          'username': username,
                          'password': pass,
                        },
                  'smb': {'type': type,
                          'username': username,
                          'password': pass,
                         },
                   }
        """
        credentials = {}  # type: Dict

        for credential in cred_tree:
            service = credential.attrib.get('service')
            credentials[service] = {}
            credentials[service]['type'] = credential.attrib.get('type')

            if service == 'ssh':
                credentials[service]['port'] = credential.attrib.get('port')

            for param in credential:
                credentials[service][param.tag] = param.text

        return credentials

    @classmethod
    def process_target_element(cls, scanner_target: Element) -> Dict:
        """ Receive an XML object with the target, ports and credentials to run
        a scan against.

        Arguments:
            Single XML target element. The target has <hosts> and <ports>
            subelements. Hosts can be a single host, a host range, a
            comma-separated host list or a network address.
            <ports> and  <credentials> are optional. Therefore each
            ospd-scanner should check for a valid ones if needed.

            Example form:

            <target>
                <hosts>192.168.0.0/24</hosts>
                <ports>22</ports>
                <credentials>
                    <credential type="up" service="ssh" port="22">
                    <username>scanuser</username>
                    <password>mypass</password>
                    </credential>
                    <credential type="up" service="smb">
                    <username>smbuser</username>
                    <password>mypass</password>
                    </credential>
                </credentials>
                <alive_test></alive_test>
                <reverse_lookup_only>1</reverse_lookup_only>
                <reverse_lookup_unify>0</reverse_lookup_unify>
            </target>

        Return:
            A Dict  hosts, port, {credentials}, exclude_hosts, options].

            Example form:

            {
                'hosts': '192.168.0.0/24',
                'port': '22',
                'credentials': {'smb': {'type': type,
                                        'port': port,
                                        'username': username,
                                        'password': pass,
                                        }
                                },

                'exclude_hosts': '',
                'finished_hosts': '',
                'options': {'alive_test': 'ALIVE_TEST_CONSIDER_ALIVE',
                            'reverse_lookup_only': '1',
                            'reverse_lookup_unify': '0',
                            },
            }
        """
        if scanner_target:
            exclude_hosts = ''
            finished_hosts = ''
            ports = ''
            hosts = None
            credentials = {}  # type: Dict
            options = {}

            for child in scanner_target:
                if child.tag == 'hosts':
                    hosts = child.text
                if child.tag == 'exclude_hosts':
                    exclude_hosts = child.text
                if child.tag == 'finished_hosts':
                    finished_hosts = child.text
                if child.tag == 'ports':
                    ports = child.text
                if child.tag == 'credentials':
                    credentials = cls.process_credentials_elements(child)
                if child.tag == 'alive_test':
                    options['alive_test'] = child.text
                if child.tag == 'reverse_lookup_unify':
                    options['reverse_lookup_unify'] = child.text
                if child.tag == 'reverse_lookup_only':
                    options['reverse_lookup_only'] = child.text

            if hosts:
                return {
                    'hosts': hosts,
                    'ports': ports,
                    'credentials': credentials,
                    'exclude_hosts': exclude_hosts,
                    'finished_hosts': finished_hosts,
                    'options': options,
                }
            else:
                raise OspdError('No target to scan')


class OspResponse:
    @staticmethod
    def create_scanner_params_xml(scanner_params: Dict[str, Any]) -> Element:
        """ Returns the OSP Daemon's scanner params in xml format. """
        scanner_params_xml = Element('scanner_params')

        for param_id, param in scanner_params.items():
            param_xml = SubElement(scanner_params_xml, 'scanner_param')

            for name, value in [('id', param_id), ('type', param['type'])]:
                param_xml.set(name, value)

            for name, value in [
                ('name', param['name']),
                ('description', param['description']),
                ('default', param['default']),
                ('mandatory', param['mandatory']),
            ]:
                elem = SubElement(param_xml, name)
                elem.text = str(value)

        return scanner_params_xml


1			# Copyright (C) 2020 Greenbone Networks GmbH
2			#
3			# SPDX-License-Identifier: GPL-2.0-or-later
4			#
5			# This program is free software; you can redistribute it and/or
6			# modify it under the terms of the GNU General Public License
7			# as published by the Free Software Foundation; either version 2
8			# of the License, or (at your option) any later version.
9			#
10			# This program is distributed in the hope that it will be useful,
11			# but WITHOUT ANY WARRANTY; without even the implied warranty of
12			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13			# GNU General Public License for more details.
14			#
15			# You should have received a copy of the GNU General Public License
16			# along with this program; if not, write to the Free Software
17			# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
18
19			""" Helper classes for parsing and creating OSP XML requests and responses
20			"""
21
22			from typing import Dict, Union, List, Any
23
24			from xml.etree.ElementTree import SubElement, Element
25
26			from ospd.errors import OspdError
27
28
29			class OspRequest:
30			@staticmethod
31			def process_vts_params(
32			scanner_vts: Element,
33			) -> Dict[str, Union[Dict, List]]:
34			""" Receive an XML object with the Vulnerability Tests an their
35			parameters to be use in a scan and return a dictionary.
36
37			@param: XML element with vt subelements. Each vt has an
38			id attribute. Optional parameters can be included
39			as vt child.
40			Example form:
41			<vt_selection>
42			<vt_single id='vt1' />
43			<vt_single id='vt2'>
44			<vt_value id='param1'>value</vt_value>
45			</vt_single>
46			<vt_group filter='family=debian'/>
47			<vt_group filter='family=general'/>
48			</vt_selection>
49
50			@return: Dictionary containing the vts attribute and subelements,
51			like the VT's id and VT's parameters.
52			Example form:
53			{'vt1': {},
54			'vt2': {'value_id': 'value'},
55			'vt_groups': ['family=debian', 'family=general']}
56			"""
57			vt_selection = {} # type: Dict
58			filters = []
59
60			for vt in scanner_vts:
61			if vt.tag == 'vt_single':
62			vt_id = vt.attrib.get('id')
63			vt_selection[vt_id] = {}
64
65			for vt_value in vt:
66			if not vt_value.attrib.get('id'):
67			raise OspdError(
68			'Invalid VT preference. No attribute id'
69			)
70
71			vt_value_id = vt_value.attrib.get('id')
72			vt_value_value = vt_value.text if vt_value.text else ''
73			vt_selection[vt_id][vt_value_id] = vt_value_value
74
75			if vt.tag == 'vt_group':
76			vts_filter = vt.attrib.get('filter', None)
77
78			if vts_filter is None:
79			raise OspdError('Invalid VT group. No filter given.')
80
81			filters.append(vts_filter)
82
83			vt_selection['vt_groups'] = filters
84
85			return vt_selection
86
87			@staticmethod
88			def process_credentials_elements(cred_tree: Element) -> Dict:
89			""" Receive an XML object with the credentials to run
90			a scan against a given target.
91
92			@param:
93			<credentials>
94			<credential type="up" service="ssh" port="22">
95			<username>scanuser</username>
96			<password>mypass</password>
97			</credential>
98			<credential type="up" service="smb">
99			<username>smbuser</username>
100			<password>mypass</password>
101			</credential>
102			</credentials>
103
104			@return: Dictionary containing the credentials for a given target.
105			Example form:
106			{'ssh': {'type': type,
107			'port': port,
108			'username': username,
109			'password': pass,
110			},
111			'smb': {'type': type,
112			'username': username,
113			'password': pass,
114			},
115			}
116			"""
117			credentials = {} # type: Dict
118
119			for credential in cred_tree:
120			service = credential.attrib.get('service')
121			credentials[service] = {}
122			credentials[service]['type'] = credential.attrib.get('type')
123
124			if service == 'ssh':
125			credentials[service]['port'] = credential.attrib.get('port')
126
127			for param in credential:
128			credentials[service][param.tag] = param.text
129
130			return credentials
131
132			@classmethod
133			def process_target_element(cls, scanner_target: Element) -> Dict:
134			""" Receive an XML object with the target, ports and credentials to run
135			a scan against.
136
137			Arguments:
138			Single XML target element. The target has <hosts> and <ports>
139			subelements. Hosts can be a single host, a host range, a
140			comma-separated host list or a network address.
141			<ports> and <credentials> are optional. Therefore each
142			ospd-scanner should check for a valid ones if needed.
143
144			Example form:
145
146			<target>
147			<hosts>192.168.0.0/24</hosts>
148			<ports>22</ports>
149			<credentials>
150			<credential type="up" service="ssh" port="22">
151			<username>scanuser</username>
152			<password>mypass</password>
153			</credential>
154			<credential type="up" service="smb">
155			<username>smbuser</username>
156			<password>mypass</password>
157			</credential>
158			</credentials>
159			<alive_test></alive_test>
160			<reverse_lookup_only>1</reverse_lookup_only>
161			<reverse_lookup_unify>0</reverse_lookup_unify>
162			</target>
163
164			Return:
165			A Dict hosts, port, {credentials}, exclude_hosts, options].
166
167			Example form:
168
169			{
170			'hosts': '192.168.0.0/24',
171			'port': '22',
172			'credentials': {'smb': {'type': type,
173			'port': port,
174			'username': username,
175			'password': pass,
176			}
177			},
178
179			'exclude_hosts': '',
180			'finished_hosts': '',
181			'options': {'alive_test': 'ALIVE_TEST_CONSIDER_ALIVE',
182			'reverse_lookup_only': '1',
183			'reverse_lookup_unify': '0',
184			},
185			}
186			"""
187			if scanner_target:
188			exclude_hosts = ''
189			finished_hosts = ''
190			ports = ''
191			hosts = None
192			credentials = {} # type: Dict
193			options = {}
194
195			for child in scanner_target:
196			if child.tag == 'hosts':
197			hosts = child.text
198			if child.tag == 'exclude_hosts':
199			exclude_hosts = child.text
200			if child.tag == 'finished_hosts':
201			finished_hosts = child.text
202			if child.tag == 'ports':
203			ports = child.text
204			if child.tag == 'credentials':
205			credentials = cls.process_credentials_elements(child)
206			if child.tag == 'alive_test':
207			options['alive_test'] = child.text
208			if child.tag == 'reverse_lookup_unify':
209			options['reverse_lookup_unify'] = child.text
210			if child.tag == 'reverse_lookup_only':
211			options['reverse_lookup_only'] = child.text
212
213			if hosts:
214			return {
215			'hosts': hosts,
216			'ports': ports,
217			'credentials': credentials,
218			'exclude_hosts': exclude_hosts,
219			'finished_hosts': finished_hosts,
220			'options': options,
221			}
222			else:
223			raise OspdError('No target to scan')
224
225
226			class OspResponse:
227			@staticmethod
228			def create_scanner_params_xml(scanner_params: Dict[str, Any]) -> Element:
229			""" Returns the OSP Daemon's scanner params in xml format. """
230			scanner_params_xml = Element('scanner_params')
231
232			for param_id, param in scanner_params.items():
233			param_xml = SubElement(scanner_params_xml, 'scanner_param')
234
235			for name, value in [('id', param_id), ('type', param['type'])]:
236			param_xml.set(name, value)
237
238			for name, value in [
239			('name', param['name']),
240			('description', param['description']),
241			('default', param['default']),
242			('mandatory', param['mandatory']),
243			]:
244			elem = SubElement(param_xml, name)
245			elem.text = str(value)
246
247			return scanner_params_xml
248

greenbone / ospd

Pull Request — master (#213)

ospd.protocol.OspRequest.process_target_element() D

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like