ospd.protocol.OspRequest.process_targets_element() - Code Metrics - Inspection of "Cleanup" - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#202)

unknown

created 2020-02-10 13:26 UTC

ospd.protocol.OspRequest.process_targets_element() D

↳ Parent: ospd.protocol

Complexity

Conditions

Size

Total Lines	94
Code Lines	37

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	12
eloc	37
nop	2
dl	0
loc	94
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 import ElementTree as et

from ospd.errors import OspdError


class OspRequest:
    @staticmethod
    def process_vts_params(
        scanner_vts: et.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: et.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_targets_element(cls, scanner_target: et.Element) -> List:
        """ Receive an XML object with the target, ports and credentials to run
        a scan against.

        @param: XML element with target subelements. Each 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:
                <targets>
                  <target>
                    <hosts>localhosts</hosts>
                    <exclude_hosts>localhost1</exclude_hosts>
                    <ports>80,443</ports>
                    <alive_test></alive_test>
                    <reverse_lookup_only>1</reverse_lookup_only>
                    <reverse_lookup_unify>0</reverse_lookup_unify>
                  </target>
                  <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>
                  </target>
                </targets>

        @return: A list of [hosts, port, {credentials}, exclude_hosts, options].
                 Example form:
                 [['localhosts', '80,43', '', 'localhosts1',
                   {'alive_test': 'ALIVE_TEST_CONSIDER_ALIVE',
                    'reverse_lookup_only': '1',
                    'reverse_lookup_unify': '0',
                   }
                  ],
                  ['192.168.0.0/24', '22', {'smb': {'type': type,
                                                    'port': port,
                                                    'username': username,
                                                    'password': pass,
                                                   }}, '', {}]]
        """

        target_list = []

        for target in scanner_target:
            exclude_hosts = ''
            finished_hosts = ''
            ports = ''
            credentials = {}  # type: Dict
            options = {}

            for child in 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:

                target_list.append(
                    [
                        hosts,
                        ports,
                        credentials,
                        exclude_hosts,
                        finished_hosts,
                        options,
                    ]
                )
            else:
                raise OspdError('No target to scan')

        return target_list


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

        for param_id, param in scanner_params.items():
            param_xml = et.SubElement(scanner_params, '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 = et.SubElement(param_xml, name)
                elem.text = str(value)

        return scanner_params


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 import ElementTree as et
25
26			from ospd.errors import OspdError
27
28
29			class OspRequest:
30			@staticmethod
31			def process_vts_params(
32			scanner_vts: et.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: et.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_targets_element(cls, scanner_target: et.Element) -> List:
134			""" Receive an XML object with the target, ports and credentials to run
135			a scan against.
136
137			@param: XML element with target subelements. Each target has <hosts>
138			and <ports> subelements. Hosts can be a single host, a host range,
139			a comma-separated host list or a network address.
140			<ports> and <credentials> are optional. Therefore each ospd-scanner
141			should check for a valid ones if needed.
142
143			Example form:
144			<targets>
145			<target>
146			<hosts>localhosts</hosts>
147			<exclude_hosts>localhost1</exclude_hosts>
148			<ports>80,443</ports>
149			<alive_test></alive_test>
150			<reverse_lookup_only>1</reverse_lookup_only>
151			<reverse_lookup_unify>0</reverse_lookup_unify>
152			</target>
153			<target>
154			<hosts>192.168.0.0/24</hosts>
155			<ports>22</ports>
156			<credentials>
157			<credential type="up" service="ssh" port="22">
158			<username>scanuser</username>
159			<password>mypass</password>
160			</credential>
161			<credential type="up" service="smb">
162			<username>smbuser</username>
163			<password>mypass</password>
164			</credential>
165			</credentials>
166			</target>
167			</targets>
168
169			@return: A list of [hosts, port, {credentials}, exclude_hosts, options].
170			Example form:
171			[['localhosts', '80,43', '', 'localhosts1',
172			{'alive_test': 'ALIVE_TEST_CONSIDER_ALIVE',
173			'reverse_lookup_only': '1',
174			'reverse_lookup_unify': '0',
175			}
176			],
177			['192.168.0.0/24', '22', {'smb': {'type': type,
178			'port': port,
179			'username': username,
180			'password': pass,
181			}}, '', {}]]
182			"""
183
184			target_list = []
185
186			for target in scanner_target:
187			exclude_hosts = ''
188			finished_hosts = ''
189			ports = ''
190			credentials = {} # type: Dict
191			options = {}
192
193			for child in target:
194			if child.tag == 'hosts':
195			hosts = child.text
196			if child.tag == 'exclude_hosts':
197			exclude_hosts = child.text
198			if child.tag == 'finished_hosts':
199			finished_hosts = child.text
200			if child.tag == 'ports':
201			ports = child.text
202			if child.tag == 'credentials':
203			credentials = cls.process_credentials_elements(child)
204			if child.tag == 'alive_test':
205			options['alive_test'] = child.text
206			if child.tag == 'reverse_lookup_unify':
207			options['reverse_lookup_unify'] = child.text
208			if child.tag == 'reverse_lookup_only':
209			options['reverse_lookup_only'] = child.text
210
211			if hosts:
			0 ignored issues – show introduced 2020-01-17 12:44 UTC by Report Bug Copy Issue Report The variable `hosts` does not seem to be defined for all execution paths. Loading history...
212			target_list.append(
213			[
214			hosts,
215			ports,
216			credentials,
217			exclude_hosts,
218			finished_hosts,
219			options,
220			]
221			)
222			else:
223			raise OspdError('No target to scan')
224
225			return target_list
226
227
228			class OspResponse:
229			@staticmethod
230			def create_scanner_params_xml(scanner_params: Dict[str, Any]) -> et.Element:
231			""" Returns the OSP Daemon's scanner params in xml format. """
232			scanner_params = et.Element('scanner_params')
233
234			for param_id, param in scanner_params.items():
235			param_xml = et.SubElement(scanner_params, 'scanner_param')
236
237			for name, value in [('id', param_id), ('type', param['type'])]:
238			param_xml.set(name, value)
239
240			for name, value in [
241			('name', param['name']),
242			('description', param['description']),
243			('default', param['default']),
244			('mandatory', param['mandatory']),
245			]:
246			elem = et.SubElement(param_xml, name)
247			elem.text = str(value)
248
249			return scanner_params
250

greenbone / ospd

Pull Request — master (#202)

ospd.protocol.OspRequest.process_targets_element() D

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like