path: root/ansible_collections/community/dns/plugins/module_utils/wsdl.py

# -*- coding: utf-8 -*-
#
# Copyright (c) 2017-2020 Felix Fontein
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later

from __future__ import (absolute_import, division, print_function)
__metaclass__ = type


from ansible.module_utils.common.text.converters import to_native
from ansible.module_utils.six import string_types

try:
    import lxml.etree
    HAS_LXML_ETREE = True
except ImportError:
    HAS_LXML_ETREE = False

from ansible_collections.community.dns.plugins.module_utils.http import (
    NetworkError,
)


class WSDLException(Exception):
    pass


class WSDLNetworkError(WSDLException):
    pass


class WSDLError(WSDLException):
    def __init__(self, origin, error_code, message):
        super(WSDLError, self).__init__('{0} ({1}): {2}'.format(origin, error_code, message))
        self.error_origin = origin
        self.error_code = error_code
        self.error_message = message


class WSDLCodingException(WSDLException):
    pass


def _split_text_namespace(node, text):
    i = text.find(':')
    if i < 0:
        return text, None
    ns = node.nsmap.get(text[:i])
    text = text[i + 1:]
    return text, ns


_NAMESPACE_ENVELOPE = 'http://schemas.xmlsoap.org/soap/envelope/'
_NAMESPACE_XSI = 'http://www.w3.org/2001/XMLSchema-instance'
_NAMESPACE_XSD = 'http://www.w3.org/2001/XMLSchema'
_NAMESPACE_XML_SOAP = 'http://xml.apache.org/xml-soap'
_NAMESPACE_XML_SOAP_ENCODING = 'http://schemas.xmlsoap.org/soap/encoding/'


def _set_type(node, type_value, namespace=None):
    if namespace is not None:
        type_value = lxml.etree.QName(namespace, type_value)
    node.set(lxml.etree.QName(_NAMESPACE_XSI, 'type').text, type_value)


def encode_wsdl(node, value):
    if value is None:
        node.set(lxml.etree.QName(_NAMESPACE_XSI, 'nil').text, 'true')
    elif isinstance(value, string_types):
        _set_type(node, 'xsd:string')
        node.text = value
    elif isinstance(value, int):
        _set_type(node, 'xsd:int')
        node.text = str(value)
    elif isinstance(value, bool):
        _set_type(node, 'xsd:boolean')
        node.text = ('true' if value else 'false')
    elif isinstance(value, dict):
        _set_type(node, 'Map', _NAMESPACE_XML_SOAP)
        for key, val in sorted(value.items()):
            child = lxml.etree.Element('item')
            ke = lxml.etree.Element('key')
            encode_wsdl(ke, key)
            child.append(ke)
            ve = lxml.etree.Element('value')
            encode_wsdl(ve, val)
            child.append(ve)
            node.append(child)
    elif isinstance(value, list):
        _set_type(node, 'SOAP-ENC:Array')
        for elt in value:
            child = lxml.etree.Element('item')
            encode_wsdl(child, elt)
            node.append(child)
    else:
        raise WSDLCodingException('Do not know how to encode {0}!'.format(type(value)))


def _decode_wsdl_array(result, node, root_ns, ids):
    for item in node:
        if item.tag != 'item':
            raise WSDLCodingException('Invalid child tag "{0}" in map!'.format(item.tag))
        result.append(decode_wsdl(item, root_ns, ids))


def decode_wsdl(node, root_ns, ids):
    href = node.get('href')
    nil = node.get(lxml.etree.QName(_NAMESPACE_XSI, 'nil'))
    id = node.get('id')
    if href is not None:
        if not href.startswith('#'):
            raise WSDLCodingException('Global reference "{0}" not supported!'.format(href))
        href = href[1:]
        if href not in ids:
            raise WSDLCodingException('ID "{0}" not yet defined!'.format(href))
        result = ids[href]
    elif nil == 'true':
        result = None
    else:
        type_with_ns = node.get(lxml.etree.QName(_NAMESPACE_XSI, 'type'))
        if type_with_ns is None:
            raise WSDLCodingException('Element "{0}" has no "xsi:type" tag!'.format(node))
        type, ns = _split_text_namespace(node, type_with_ns)
        if ns is None:
            raise WSDLCodingException('Cannot find namespace for "{0}"!'.format(type_with_ns))
        if ns == _NAMESPACE_XSD:
            if type == 'boolean':
                if node.text == 'true':
                    result = True
                elif node.text == 'false':
                    result = False
                else:
                    raise WSDLCodingException('Invalid value for boolean: "{0}"'.format(node.text))
            elif type == 'int':
                result = int(node.text)
            elif type == 'string':
                result = node.text
            else:
                raise WSDLCodingException('Unknown XSD type "{0}"!'.format(type))
        elif ns == _NAMESPACE_XML_SOAP:
            if type == 'Map':
                result = dict()
                if id is not None:
                    ids[id] = result
                for item in node:
                    if item.tag != 'item':
                        raise WSDLCodingException('Invalid child tag "{0}" in map!'.format(item.tag))
                    key = item.find('key')
                    if key is None:
                        raise WSDLCodingException('Cannot find key for "{0}"!'.format(item))
                    key = decode_wsdl(key, root_ns, ids)
                    value = item.find('value')
                    if value is None:
                        raise WSDLCodingException('Cannot find value for "{0}"!'.format(item))
                    value = decode_wsdl(value, root_ns, ids)
                    result[key] = value
                return result
            else:
                raise WSDLCodingException('Unknown XSD type "{0}"!'.format(type))
        elif ns == _NAMESPACE_XML_SOAP_ENCODING:
            if type == 'Array':
                result = []
                if id is not None:
                    ids[id] = result
                _decode_wsdl_array(result, node, root_ns, ids)
            else:
                raise WSDLCodingException('Unknown XSD type "{0}"!'.format(type))
        elif ns == root_ns:
            array_type = node.get(lxml.etree.QName(_NAMESPACE_XML_SOAP_ENCODING, 'arrayType'))
            if array_type is not None:
                result = []
                if id is not None:
                    ids[id] = result
                _decode_wsdl_array(result, node, root_ns, ids)
            else:
                result = dict()
                if id is not None:
                    ids[id] = result
                for item in node:
                    result[item.tag] = decode_wsdl(item, root_ns, ids)
        else:
            raise WSDLCodingException('Unknown type namespace "{0}" (with type "{1}")!'.format(ns, type))
    if id is not None:
        ids[id] = result
    return result


class Parser(object):
    def _parse(self, result, node, where):
        for child in node:
            tag = lxml.etree.QName(child.tag)
            if tag.namespace != self._api:
                raise WSDLCodingException('Cannot interpret {0} item of type "{1}"!'.format(where, tag))
            for res in child.iter('return'):
                result[tag.localname] = decode_wsdl(res, self._api, {})

    def __init__(self, api, root):
        self._main_ns = _NAMESPACE_ENVELOPE
        self._api = api
        self._root = root
        for fault in self._root.iter(lxml.etree.QName(self._main_ns, 'Fault').text):
            fault_code = fault.find('faultcode')
            fault_code_val = None
            fault_string = fault.find('faultstring')
            origin = 'server'
            if fault_code is not None and fault_code.text:
                code, code_ns = _split_text_namespace(fault, fault_code.text)
                fault_code_val = code
                if code_ns == self._main_ns:
                    origin = code.lower()
            if fault_string is not None and fault_string.text:
                raise WSDLError(origin, fault_code_val, fault_string.text)
            raise WSDLError(origin, fault_code_val, lxml.etree.tostring(fault).decode('utf-8'))
        self._header = dict()
        self._body = dict()
        for header in self._root.iter(lxml.etree.QName(self._main_ns, 'Header').text):
            self._parse(self._header, header, 'header')
        for body in self._root.iter(lxml.etree.QName(self._main_ns, 'Body').text):
            self._parse(self._body, body, 'body')

    def get_header(self, header):
        return self._header[header]

    def get_result(self, body):
        return self._body[body]

    def __str__(self):
        return 'header={0}, body={1}'.format(self._header, self._body)

    def __repr__(self):
        return '''<?xml version='1.0' encoding='utf-8'?>''' + '\n' + lxml.etree.tostring(self._root, pretty_print=True).decode('utf-8')


class Composer(object):
    @staticmethod
    def _create(tag, namespace=None, **kwarg):
        if namespace:
            return lxml.etree.Element(lxml.etree.QName(namespace, tag), **kwarg)
        else:
            return lxml.etree.Element(tag, **kwarg)

    def __str__(self):
        return '''<?xml version='1.0' encoding='utf-8'?>''' + '\n' + lxml.etree.tostring(self._root, pretty_print=True).decode('utf-8')

    def _create_envelope(self, tag, **kwarg):
        return self._create(tag, self._main_ns, **kwarg)

    def __init__(self, http_helper, api, namespaces=None):
        self._http_helper = http_helper
        self._main_ns = _NAMESPACE_ENVELOPE
        self._api = api
        # Compose basic document
        all_namespaces = {
            'SOAP-ENV': _NAMESPACE_ENVELOPE,
            'xsd': _NAMESPACE_XSD,
            'xsi': _NAMESPACE_XSI,
            'ns2': 'auth',
            'SOAP-ENC': _NAMESPACE_XML_SOAP_ENCODING,
        }
        if namespaces is not None:
            all_namespaces.update(namespaces)
        self._root = self._create_envelope('Envelope', nsmap=all_namespaces)
        self._root.set(lxml.etree.QName(self._main_ns, 'encodingStyle').text, _NAMESPACE_XML_SOAP_ENCODING)
        self._header = self._create_envelope('Header')
        self._root.append(self._header)
        self._body = self._create_envelope('Body')
        self._root.append(self._body)
        self._command = None

    def add_auth(self, username, password):
        auth = self._create('authenticate', 'auth')
        user = self._create('UserName')
        user.text = username
        auth.append(user)
        pw = self._create('Password')
        pw.text = password
        auth.append(pw)
        self._header.append(auth)

    def add_simple_command(self, command, **args):
        self._command = command
        command = self._create(command, self._api)
        for arg, value in args.items():
            arg = self._create(arg)
            encode_wsdl(arg, value)
            command.append(arg)
        self._body.append(command)

    def execute(self, debug=False):
        payload = b'''<?xml version='1.0' encoding='utf-8'?>''' + b'\n' + lxml.etree.tostring(self._root) + b'\n'
        try:
            headers = {
                'Content-Type': 'text/xml; charset=utf-8',
                'Content-Length': str(len(payload)),
            }
            if self._command:
                headers['SOAPAction'] = '"{0}#{1}"'.format(self._api, self._command)
            result, info = self._http_helper.fetch_url(self._api, data=payload, method='POST', timeout=300, headers=headers)
            code = info['status']
        except NetworkError as e:
            raise WSDLNetworkError(to_native(e))
        # if debug:
        #     q.q('Result: {0}, content: {1}'.format(code, result.decode('utf-8')))
        if code < 200 or code >= 300:
            Parser(self._api, lxml.etree.fromstring(result))
            raise WSDLError('server', 'Error {0} while executing WSDL command:\n{1}'.format(code, result.decode('utf-8')))
        return Parser(self._api, lxml.etree.fromstring(result))