ospd.xml.XmlStringHelper.add_element() - Code Metrics - Inspection of "Improve handling of dead hosts and improve scan pr..." - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#266)

by Juan José

created 2020-05-11 12:31 UTC

ospd.xml.XmlStringHelper.add_element() B

↳ Parent: ospd.xml

Complexity

Conditions

Size

Total Lines	35
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	17
nop	4
dl	0
loc	35
rs	8
c	0
b	0
f	0

# 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/>.

""" OSP XML utils class.
"""

import re

from typing import List, Dict, Any, Union

from xml.sax.saxutils import escape, quoteattr
from xml.etree.ElementTree import tostring, Element

from ospd.misc import ResultType


r = re.compile(  # pylint: disable=invalid-name
    r'(.*?)(?:([^\x09\x0A\x0D\x20-\x7E\x85\xA0-\xFF'
    + r'\u0100-\uD7FF\uE000-\uFDCF\uFDE0-\uFFFD])|([\n])|$)'
)


def split_invalid_xml(result_text: str) -> Union[List[Union[str, int]], str]:
    """ Search for occurrence of non printable chars and replace them
    with the integer representation the Unicode code. The original string
    is splitted where a non printable char is found.
    """
    splitted_string = []

    def replacer(match):
        regex_g1 = match.group(1)
        if len(regex_g1) > 0:
            splitted_string.append(regex_g1)
        regex_g2 = match.group(2)
        if regex_g2 is not None:
            splitted_string.append(ord(regex_g2))
        regex_g3 = match.group(3)
        if regex_g3 is not None:
            splitted_string.append(regex_g3)
        return ""

    re.sub(r, replacer, result_text)
    return splitted_string


def escape_ctrl_chars(result_text):
    """ Replace non printable chars in result_text with an hexa code
    in string format.
    """
    escaped_str = ''
    for fragment in split_invalid_xml(result_text):
        if isinstance(fragment, int):
            escaped_str += '\\x%04X' % fragment
        else:
            escaped_str += fragment

    return escaped_str


def get_result_xml(result):
    """ Formats a scan result to XML format.

    Arguments:
        result (dict): Dictionary with a scan result.

    Return:
        Result as xml element object.
    """

    result_xml = Element('result')
    for name, value in [
        ('name', result['name']),
        ('type', ResultType.get_str(result['type'])),
        ('severity', result['severity']),
        ('host', result['host']),
        ('hostname', result['hostname']),
        ('test_id', result['test_id']),
        ('port', result['port']),
        ('qod', result['qod']),
    ]:
        result_xml.set(name, escape(str(value)))
    if result['value'] is not None:
        result_xml.text = escape_ctrl_chars(result['value'])

    return result_xml


def get_progress_xml(progress: Dict[str, int]):
    """ Formats a scan progress to XML format.

    Arguments:
        progress (dict): Dictionary with a scan progress.

    Return:
        Progress as xml element object.
    """

    progress_xml = Element('progress')
    for progress_item, value in progress.items():
        elem = None
        if progress_item == 'current_hosts':
            for host, h_progress in value.items():
                elem = Element('host')
                elem.set('name', host)
                elem.text = str(h_progress)
                progress_xml.append(elem)
        else:
            elem = Element(progress_item)
            elem.text = str(value)
            progress_xml.append(elem)
    return progress_xml


def simple_response_str(
    command: str,
    status: int,
    status_text: str,
    content: Union[str, Element, List[str], List[Element]] = "",
) -> bytes:
    """ Creates an OSP response XML string.

    Arguments:
        command (str): OSP Command to respond to.
        status (int): Status of the response.
        status_text (str): Status text of the response.
        content (str): Text part of the response XML element.

    Return:
        String of response in xml format.
    """
    response = Element('%s_response' % command)

    for name, value in [('status', str(status)), ('status_text', status_text)]:
        response.set(name, escape(str(value)))

    if isinstance(content, list):
        for elem in content:
            if isinstance(elem, Element):
                response.append(elem)
    elif isinstance(content, Element):
        response.append(content)
    elif content is not None:
        response.text = escape_ctrl_chars(content)

    return tostring(response, encoding='utf-8')


def get_elements_from_dict(data: Dict[str, Any]) -> List[Element]:
    """ Creates a list of etree elements from a dictionary

    Args:
        Dictionary of tags and their elements.

    Return:
        List of xml elements.
    """

    responses = []

    for tag, value in data.items():
        elem = Element(tag)

        if isinstance(value, dict):
            for val in get_elements_from_dict(value):
                elem.append(val)
        elif isinstance(value, list):
            elem.text = ', '.join(value)
        elif value is not None:
            elem.text = escape_ctrl_chars(value)

        responses.append(elem)

    return responses


def elements_as_text(
    elements: Dict[str, Union[str, Dict]], indent: int = 2
) -> str:
    """ Returns the elements dictionary as formatted plain text. """

    text = ""
    for elename, eledesc in elements.items():
        if isinstance(eledesc, dict):
            desc_txt = elements_as_text(eledesc, indent + 2)
            desc_txt = ''.join(['\n', desc_txt])
        elif isinstance(eledesc, str):
            desc_txt = ''.join([eledesc, '\n'])
        else:
            assert False, "Only string or dictionary"

        ele_txt = "\t{0}{1: <22} {2}".format(' ' * indent, elename, desc_txt)


        text = ''.join([text, ele_txt])

    return text


class XmlStringHelper:
    """ Class with methods to help the creation of a xml object in
    string format.
    """

    def create_element(self, elem_name: str, end: bool = False) -> bytes:
        """ Get a name and create the open element of an entity.

        Arguments:
            elem_name (str): The name of the tag element.
            end (bool): Create a initial tag if False, otherwise the end tag.

        Return:
            Encoded string representing a part of an xml element.
        """
        if end:
            ret = "</%s>" % elem_name
        else:
            ret = "<%s>" % elem_name

        return ret.encode('utf-8')

    def create_response(self, command: str, end: bool = False) -> bytes:
        """ Create or end an xml response.

        Arguments:
            command (str): The name of the command for the response element.
            end (bool): Create a initial tag if False, otherwise the end tag.

        Return:
            Encoded string representing a part of an xml element.
        """
        if not command:
            return

        if end:
            return ('</%s_response>' % command).encode('utf-8')

        return ('<%s_response status="200" status_text="OK">' % command).encode(
            'utf-8'
        )

    def add_element(
        self,
        content: Union[Element, str, list],
        xml_str: bytes = None,
        end: bool = False,
    ) -> bytes:
        """Create the initial or ending tag for a subelement, or add
        one or many xml elements

        Arguments:
            content (Element, str, list): Content to add.
            xml_str (bytes): Initial string where content to be added to.
            end (bool): Create a initial tag if False, otherwise the end tag.
                        It will be added to the xml_str.

        Return:
            Encoded string representing a part of an xml element.
        """

        if not xml_str:
            xml_str = b''

        if content:
            if isinstance(content, list):
                for elem in content:
                    xml_str = xml_str + tostring(elem, encoding='utf-8')
            elif isinstance(content, Element):
                xml_str = xml_str + tostring(content, encoding='utf-8')
            else:
                if end:
                    xml_str = xml_str + self.create_element(content, False)
                else:
                    xml_str = xml_str + self.create_element(content)

        return xml_str

    def add_attr(
        self, tag: bytes, attribute: str, value: Union[str, int] = None
    ) -> bytes:
        """ Add an attribute to the beginning tag of an xml element.
        Arguments:
            tag (bytes): Tag to add the attribute to.
            attribute (str): Attribute name
            value (str): Attribute value
        Return:
            Tag in encoded string format with the given attribute
        """
        if not tag:
            return None

        if not attribute:
            return tag

        if not value:
            value = ''

        return tag[:-1] + (
            " %s=%s>" % (attribute, quoteattr(str(value)))
        ).encode('utf-8')


1			# Copyright (C) 2014-2020 Greenbone Networks GmbH
2			#
3			# SPDX-License-Identifier: AGPL-3.0-or-later
4			#
5			# This program is free software: you can redistribute it and/or modify
6			# it under the terms of the GNU Affero General Public License as
7			# published by the Free Software Foundation, either version 3 of the
8			# 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 Affero General Public License for more details.
14			#
15			# You should have received a copy of the GNU Affero General Public License
16			# along with this program. If not, see <http://www.gnu.org/licenses/>.
17
18			""" OSP XML utils class.
19			"""
20
21			import re
22
23			from typing import List, Dict, Any, Union
24
25			from xml.sax.saxutils import escape, quoteattr
26			from xml.etree.ElementTree import tostring, Element
27
28			from ospd.misc import ResultType
29
30
31			r = re.compile( # pylint: disable=invalid-name
32			r'(.*?)(?:([^\x09\x0A\x0D\x20-\x7E\x85\xA0-\xFF'
33			+ r'\u0100-\uD7FF\uE000-\uFDCF\uFDE0-\uFFFD])\|([\n])\|$)'
34			)
35
36
37			def split_invalid_xml(result_text: str) -> Union[List[Union[str, int]], str]:
38			""" Search for occurrence of non printable chars and replace them
39			with the integer representation the Unicode code. The original string
40			is splitted where a non printable char is found.
41			"""
42			splitted_string = []
43
44			def replacer(match):
45			regex_g1 = match.group(1)
46			if len(regex_g1) > 0:
47			splitted_string.append(regex_g1)
48			regex_g2 = match.group(2)
49			if regex_g2 is not None:
50			splitted_string.append(ord(regex_g2))
51			regex_g3 = match.group(3)
52			if regex_g3 is not None:
53			splitted_string.append(regex_g3)
54			return ""
55
56			re.sub(r, replacer, result_text)
57			return splitted_string
58
59
60			def escape_ctrl_chars(result_text):
61			""" Replace non printable chars in result_text with an hexa code
62			in string format.
63			"""
64			escaped_str = ''
65			for fragment in split_invalid_xml(result_text):
66			if isinstance(fragment, int):
67			escaped_str += '\\x%04X' % fragment
68			else:
69			escaped_str += fragment
70
71			return escaped_str
72
73
74			def get_result_xml(result):
75			""" Formats a scan result to XML format.
76
77			Arguments:
78			result (dict): Dictionary with a scan result.
79
80			Return:
81			Result as xml element object.
82			"""
83
84			result_xml = Element('result')
85			for name, value in [
86			('name', result['name']),
87			('type', ResultType.get_str(result['type'])),
88			('severity', result['severity']),
89			('host', result['host']),
90			('hostname', result['hostname']),
91			('test_id', result['test_id']),
92			('port', result['port']),
93			('qod', result['qod']),
94			]:
95			result_xml.set(name, escape(str(value)))
96			if result['value'] is not None:
97			result_xml.text = escape_ctrl_chars(result['value'])
98
99			return result_xml
100
101
102			def get_progress_xml(progress: Dict[str, int]):
103			""" Formats a scan progress to XML format.
104
105			Arguments:
106			progress (dict): Dictionary with a scan progress.
107
108			Return:
109			Progress as xml element object.
110			"""
111
112			progress_xml = Element('progress')
113			for progress_item, value in progress.items():
114			elem = None
115			if progress_item == 'current_hosts':
116			for host, h_progress in value.items():
117			elem = Element('host')
118			elem.set('name', host)
119			elem.text = str(h_progress)
120			progress_xml.append(elem)
121			else:
122			elem = Element(progress_item)
123			elem.text = str(value)
124			progress_xml.append(elem)
125			return progress_xml
126
127
128			def simple_response_str(
129			command: str,
130			status: int,
131			status_text: str,
132			content: Union[str, Element, List[str], List[Element]] = "",
133			) -> bytes:
134			""" Creates an OSP response XML string.
135
136			Arguments:
137			command (str): OSP Command to respond to.
138			status (int): Status of the response.
139			status_text (str): Status text of the response.
140			content (str): Text part of the response XML element.
141
142			Return:
143			String of response in xml format.
144			"""
145			response = Element('%s_response' % command)
146
147			for name, value in [('status', str(status)), ('status_text', status_text)]:
148			response.set(name, escape(str(value)))
149
150			if isinstance(content, list):
151			for elem in content:
152			if isinstance(elem, Element):
153			response.append(elem)
154			elif isinstance(content, Element):
155			response.append(content)
156			elif content is not None:
157			response.text = escape_ctrl_chars(content)
158
159			return tostring(response, encoding='utf-8')
160
161
162			def get_elements_from_dict(data: Dict[str, Any]) -> List[Element]:
163			""" Creates a list of etree elements from a dictionary
164
165			Args:
166			Dictionary of tags and their elements.
167
168			Return:
169			List of xml elements.
170			"""
171
172			responses = []
173
174			for tag, value in data.items():
175			elem = Element(tag)
176
177			if isinstance(value, dict):
178			for val in get_elements_from_dict(value):
179			elem.append(val)
180			elif isinstance(value, list):
181			elem.text = ', '.join(value)
182			elif value is not None:
183			elem.text = escape_ctrl_chars(value)
184
185			responses.append(elem)
186
187			return responses
188
189
190			def elements_as_text(
191			elements: Dict[str, Union[str, Dict]], indent: int = 2
192			) -> str:
193			""" Returns the elements dictionary as formatted plain text. """
194
195			text = ""
196			for elename, eledesc in elements.items():
197			if isinstance(eledesc, dict):
198			desc_txt = elements_as_text(eledesc, indent + 2)
199			desc_txt = ''.join(['\n', desc_txt])
200			elif isinstance(eledesc, str):
201			desc_txt = ''.join([eledesc, '\n'])
202			else:
203			assert False, "Only string or dictionary"
204
205			ele_txt = "\t{0}{1: <22} {2}".format(' ' * indent, elename, desc_txt)
			0 ignored issues – show introduced 2020-02-10 13:13 UTC by Report Bug Copy Issue Report The variable `desc_txt` does not seem to be defined for all execution paths. Loading history...
206
207			text = ''.join([text, ele_txt])
208
209			return text
210
211
212			class XmlStringHelper:
213			""" Class with methods to help the creation of a xml object in
214			string format.
215			"""
216
217			def create_element(self, elem_name: str, end: bool = False) -> bytes:
218			""" Get a name and create the open element of an entity.
219
220			Arguments:
221			elem_name (str): The name of the tag element.
222			end (bool): Create a initial tag if False, otherwise the end tag.
223
224			Return:
225			Encoded string representing a part of an xml element.
226			"""
227			if end:
228			ret = "</%s>" % elem_name
229			else:
230			ret = "<%s>" % elem_name
231
232			return ret.encode('utf-8')
233
234			def create_response(self, command: str, end: bool = False) -> bytes:
235			""" Create or end an xml response.
236
237			Arguments:
238			command (str): The name of the command for the response element.
239			end (bool): Create a initial tag if False, otherwise the end tag.
240
241			Return:
242			Encoded string representing a part of an xml element.
243			"""
244			if not command:
245			return
246
247			if end:
248			return ('</%s_response>' % command).encode('utf-8')
249
250			return ('<%s_response status="200" status_text="OK">' % command).encode(
251			'utf-8'
252			)
253
254			def add_element(
255			self,
256			content: Union[Element, str, list],
257			xml_str: bytes = None,
258			end: bool = False,
259			) -> bytes:
260			"""Create the initial or ending tag for a subelement, or add
261			one or many xml elements
262
263			Arguments:
264			content (Element, str, list): Content to add.
265			xml_str (bytes): Initial string where content to be added to.
266			end (bool): Create a initial tag if False, otherwise the end tag.
267			It will be added to the xml_str.
268
269			Return:
270			Encoded string representing a part of an xml element.
271			"""
272
273			if not xml_str:
274			xml_str = b''
275
276			if content:
277			if isinstance(content, list):
278			for elem in content:
279			xml_str = xml_str + tostring(elem, encoding='utf-8')
280			elif isinstance(content, Element):
281			xml_str = xml_str + tostring(content, encoding='utf-8')
282			else:
283			if end:
284			xml_str = xml_str + self.create_element(content, False)
285			else:
286			xml_str = xml_str + self.create_element(content)
287
288			return xml_str
289
290			def add_attr(
291			self, tag: bytes, attribute: str, value: Union[str, int] = None
292			) -> bytes:
293			""" Add an attribute to the beginning tag of an xml element.
294			Arguments:
295			tag (bytes): Tag to add the attribute to.
296			attribute (str): Attribute name
297			value (str): Attribute value
298			Return:
299			Tag in encoded string format with the given attribute
300			"""
301			if not tag:
302			return None
303
304			if not attribute:
305			return tag
306
307			if not value:
308			value = ''
309
310			return tag[:-1] + (
311			" %s=%s>" % (attribute, quoteattr(str(value)))
312			).encode('utf-8')
313

greenbone / ospd

Pull Request — master (#266)

ospd.xml.XmlStringHelper.add_element() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like