Produits Entr'ouvert » Passerelle

# Generated by Django 2.2.24 on 2022-01-19 16:37

from django.db import migrations, models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0029_auto_20210202_1627'),

    ]

    operations = [

        migrations.CreateModel(

            name='SOAPConnector',

            fields=[

                (

                    'id',

                    models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),

                ),

                ('title', models.CharField(max_length=50, verbose_name='Title')),

                ('slug', models.SlugField(unique=True, verbose_name='Identifier')),

                ('description', models.TextField(verbose_name='Description')),

                (

                    'basic_auth_username',

                    models.CharField(

                        blank=True, max_length=128, verbose_name='Basic authentication username'

                    ),

                ),

                (

                    'basic_auth_password',

                    models.CharField(

                        blank=True, max_length=128, verbose_name='Basic authentication password'

                    ),

                ),

                (

                    'client_certificate',

                    models.FileField(

                        blank=True, null=True, upload_to='', verbose_name='TLS client certificate'

                    ),

                ),

                (

                    'trusted_certificate_authorities',

                    models.FileField(blank=True, null=True, upload_to='', verbose_name='TLS trusted CAs'),

                ),

                (

                    'verify_cert',

                    models.BooleanField(blank=True, default=True, verbose_name='TLS verify certificates'),

                ),

                (

                    'http_proxy',

                    models.CharField(blank=True, max_length=128, verbose_name='HTTP and HTTPS proxy'),

                ),

                (

                    'wsdl_url',

                    models.URLField(

                        help_text='URL of the WSDL file', max_length=400, verbose_name='WSDL URL'

                    ),

                ),

                (

                    'zeep_strict',

                    models.BooleanField(default=True, verbose_name='Be strict with returned XML'),

                ),

                (

                    'zeep_xsd_ignore_sequence_order',

                    models.BooleanField(default=False, verbose_name='Ignore sequence order'),

                ),

                (

                    'users',

                    models.ManyToManyField(

                        blank=True,

                        related_name='_soapconnector_users_+',

                        related_query_name='+',

                        to='base.ApiUser',

                    ),

                ),

            ],

            options={

                'verbose_name': 'SOAP connector',

            },

        ),

    ]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2019  Entr'ouvert

#

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

import collections

import zeep

import zeep.exceptions

import zeep.helpers

import zeep.xsd

from django.db import models

from django.utils.functional import cached_property

from django.utils.translation import ugettext_lazy as _

from passerelle.base.models import BaseResource, HTTPResource

from passerelle.utils.api import endpoint

from passerelle.utils.json import unflatten

class SOAPConnector(BaseResource, HTTPResource):

    wsdl_url = models.URLField(

        max_length=400, verbose_name=_('WSDL URL'), help_text=_('URL of the WSDL file')

    )

    zeep_strict = models.BooleanField(default=False, verbose_name=_('Be strict with returned XML'))

    zeep_xsd_ignore_sequence_order = models.BooleanField(

        default=True, verbose_name=_('Ignore sequence order')

    )

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = _('SOAP connector')

    @classmethod

    def get_manager_form_class(cls, **kwargs):

        form_class = super().get_manager_form_class(**kwargs)

        fields = list(form_class.base_fields.items())

        form_class.base_fields = collections.OrderedDict(fields[:2] + fields[-3:] + fields[2:-3])

        return form_class

    @cached_property

    def client(self):

        return self.soap_client(

            wsdl_url=self.wsdl_url,

            settings=zeep.Settings(

                strict=self.zeep_strict, xsd_ignore_sequence_order=self.zeep_xsd_ignore_sequence_order

            ),

        )

    @endpoint(

        methods=['post'],

        perm='can_access',

        name='method',

        pattern=r'^(?P<method_name>\w+)/$',

        example_pattern='method_name/',

        description_get=_('Call a SOAP method'),

        description_post=_('Call a SOAP method'),

        post_json_schema={'type': 'object'},

    )

    def method(self, request, method_name, post_data=None, **kwargs):

        def jsonify(data):

            if isinstance(data, (dict, collections.OrderedDict)):

                # ignore _raw_elements, zeep put there nodes not maching the

                # XSD when strict parsing is disabled.

                return {

                    jsonify(k): jsonify(v)

                    for k, v in data.items()

                    if (self.zeep_strict or k != '_raw_elements')

                }

            elif isinstance(data, (list, tuple, collections.deque)):

                return [jsonify(item) for item in data]

            else:

                return data

        payload = {}

        for k in request.GET:

            if k == 'raise':

                continue

            value = request.GET.getlist(k)

            if len(value) > 1:

                payload[k] = value

            else:

                payload[k] = value[0]

        payload.update(post_data or {})

        payload = unflatten(payload)

        try:

            soap_response = getattr(self.client.service, method_name)(**payload)

        except zeep.exceptions.ValidationError as e:

            e.status_code = 400

            raise e

        serialized = zeep.helpers.serialize_object(soap_response)

        json_response = jsonify(serialized)

        return {'err': 0, 'data': json_response}

    method.endpoint_info.methods.append('get')

    def get_endpoints_infos(self):

        endpoints = super().get_endpoints_infos()

        for name, input_schema, output_schema in self.operations_and_schemas:

            kwargs = dict(

                name='method',

                pattern=f'{name}/',

                example_pattern=f'{name}/',

                description=f'Method {name}',

                json_schema_response={

                    'type': 'object',

                    'properties': collections.OrderedDict(

                        [

                            ('err', {'type': 'integer'}),

                            ('data', output_schema),

                        ]

                    ),

                },

            )

            if input_schema:

                kwargs['post_json_schema'] = input_schema

            endpoints.append(endpoint(**kwargs))

            endpoints[-1].object = self

            endpoints[-1].func = lambda request: None

            if input_schema:

                endpoints[-1].http_method = 'post'

            else:

                endpoints[-1].http_method = 'get'

        return endpoints

    @property

    def operations_and_schemas(self):

        operations = self.client.service._binding._operations

        for name in operations:

            operation = operations[name]

            input_type = operation.input.body.type

            output_type = operation.output.body.type

            input_schema = self.type2schema(input_type, keep_root=True)

            output_schema = self.type2schema(output_type, compress=True)

            yield name, input_schema, output_schema

    def type2schema(self, xsd_type, keep_root=False, compress=False):

        # simplify schema: when a type contains a unique element, it will try

        # to match any dict or list with it on input and will flatten the

        # schema on output.

        if (

            isinstance(xsd_type, zeep.xsd.ComplexType)

            and len(xsd_type.elements) == 1

            and not keep_root

            and compress

        ):

            if xsd_type.elements[0][1].max_occurs != 1:

                return {

                    'type': 'array',

                    'items': self.type2schema(xsd_type.elements[0][1].type, compress=compress),

                }

            return self.type2schema(xsd_type.elements[0][1].type, compress=compress)

        if isinstance(xsd_type, zeep.xsd.ComplexType):

            properties = collections.OrderedDict()

            schema = {

                'type': 'object',

                'properties': properties,

            }

            for key, element in xsd_type.elements:

                if element.min_occurs > 0:

                    schema.setdefault('required', []).append(key)

                element_schema = self.type2schema(element.type, compress=compress)

                if element.max_occurs == 'unbounded' or element.max_occurs > 1:

                    element_schema = {'type': 'array', 'items': element_schema}

                properties[key] = element_schema

            if not properties:

                return None

            return schema

        if isinstance(xsd_type, zeep.xsd.BuiltinType):

            return {'type': 'string'}

        return f'!!! UNKNOWN TYPE {xsd_type} !!!'

{% extends "passerelle/manage/service_view.html" %}

{% load i18n passerelle %}

    'passerelle.apps.soap',

import urllib.parse

import utils

from passerelle.apps.soap.models import SOAPConnector

class SOAP11:

    VERSION = '1.1'

    ENDPOINT_URL = 'https://www.examples.com/SayHello/'

    WSDL_CONTENT = '''\

<definitions name = "HelloService"

   targetNamespace = "http://www.examples.com/wsdl/HelloService.wsdl"

   xmlns = "http://schemas.xmlsoap.org/wsdl/"

   xmlns:soap = "http://schemas.xmlsoap.org/wsdl/soap/"

   xmlns:tns = "http://www.examples.com/wsdl/HelloService.wsdl"

   xmlns:xsd = "http://www.w3.org/2001/XMLSchema">

   <types>

     <schema targetNamespace="http://www.examples.com/wsdl/HelloService.wsdl" xmlns="http://www.w3.org/2001/XMLSchema">

      <element name="firstName">

        <complexType name="listofstring">

          <sequence>

            <element name="string" type="string" maxOccurs="unbounded"/>

          </sequence>

        </complexType>

      </element>

     </schema>

   </types>

   <message name = "SayHelloRequest">

      <part name = "firstName" element="tns:firstName"/>

      <part name = "lastName" type = "xsd:string"/>

   </message>

   <message name = "SayHelloResponse">

      <part name = "greeting" type = "xsd:string"/>

   </message>

   <portType name = "Hello_PortType">

      <operation name = "sayHello">

         <input message = "tns:SayHelloRequest"/>

         <output message = "tns:SayHelloResponse"/>

      </operation>

   </portType>

   <binding name = "Hello_Binding" type = "tns:Hello_PortType">

      <soap:binding style = "rpc"

         transport = "http://schemas.xmlsoap.org/soap/http"/>

      <operation name = "sayHello">

         <soap:operation soapAction = "sayHello"/>

         <input>

            <soap:body

               encodingStyle = "http://schemas.xmlsoap.org/soap/encoding/"

               namespace = "urn:examples:helloservice"

               use = "encoded"/>

         </input>

         <output>

            <soap:body

               encodingStyle = "http://schemas.xmlsoap.org/soap/encoding/"

               namespace = "urn:examples:helloservice"

               use = "encoded"/>

         </output>

      </operation>

   </binding>

   <service name = "Hello_Service">

      <documentation>WSDL File for HelloService</documentation>

      <port binding = "tns:Hello_Binding" name = "Hello_Port">

         <soap:address

            location = "http://www.examples.com/SayHello/" />

      </port>

   </service>

</definitions>'''

    WSDL_URL = 'https://example.com/service.wsdl'

    SOAP_RESPONSE = '''\

<?xml version="1.0" encoding="utf-8"?>

<soap:Envelope xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:xsd="http://www.w3.org/2001/XMLSchema"

xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">

   <soap:Body>

     <SayHelloResponse xmlns="urn:examples:helloservice">

       <greeting>Hello John!</greeting>

     </SayHelloResponse>

   </soap:Body>

</soap:Envelope>'''

    INPUT_SCHEMA = {

        'properties': {

            'firstName': {

                'properties': {

                    'string': {'items': {'type': 'string'}, 'type': 'array'},

                },

                'required': ['string'],

                'type': 'object',

            },

            'lastName': {'type': 'string'},

        },

        'required': ['firstName', 'lastName'],

        'type': 'object',

    }

    OUTPUT_SCHEMA = {'type': 'string'}

    INPUT_DATA = {

        'firstName/string/0': 'John',

        'firstName/string/1': 'Bill',

        'lastName': 'Doe',

    }

class SOAP12(SOAP11):

    VERSION = '1.2'

    ENDPOINT_URL = 'https://www.examples.com/SayHello/'

    WSDL_CONTENT = f'''\

<?xml version="1.0"?>

<wsdl:definitions xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/"

  xmlns:soap12="http://schemas.xmlsoap.org/wsdl/soap12/"

  xmlns:tns="urn:examples:helloservice"

  targetNamespace="urn:examples:helloservice">

  <wsdl:types>

    <xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema"

      xmlns:tns="urn:examples:helloservice"

      targetNamespace="urn:examples:helloservice">

      <xsd:element name="sayHello">

        <xsd:complexType>

          <xsd:sequence>

            <xsd:element name="firstName" type="xsd:string" maxOccurs="unbounded"/>

            <xsd:element name="lastName" type="xsd:string"/>

          </xsd:sequence>

        </xsd:complexType>

      </xsd:element>

      <xsd:element name="sayHelloResponse">

        <xsd:complexType>

          <xsd:sequence>

            <xsd:element name="greeting" type="xsd:string"/>

          </xsd:sequence>

        </xsd:complexType>

      </xsd:element>

    </xsd:schema>

  </wsdl:types>

  <wsdl:message name="sayHello">

    <wsdl:part name="sayHelloInputPart" element="tns:sayHello"/>

  </wsdl:message>

  <wsdl:message name="sayHelloResponse">

    <wsdl:part name="sayHelloOutputPart" element="tns:sayHelloResponse"/>

  </wsdl:message>

  <wsdl:portType name="sayHelloPortType">

    <wsdl:operation name="sayHello">

      <wsdl:input name="sayHello" message="tns:sayHello"/>

      <wsdl:output name="sayHelloResponse" message="tns:sayHelloResponse"/>

    </wsdl:operation>

  </wsdl:portType>

  <wsdl:binding name="sayHelloBinding"

    type="tns:sayHelloPortType">

    <soap12:binding style="document" transport="http://schemas.xmlsoap.org/soap/http"/>

    <wsdl:operation name="sayHello">

      <soap12:operation style="document"/>

      <wsdl:input name="sayHello">

        <soap12:body use="literal"/>

      </wsdl:input>

      <wsdl:output name="sayHelloResponse">

        <soap12:body use="literal"/>

      </wsdl:output>

    </wsdl:operation>

  </wsdl:binding>

  <wsdl:service name="sayHelloService">

    <wsdl:port name="sayHelloPort"

      binding="tns:sayHelloBinding">

      <soap12:address location="{ENDPOINT_URL}"/>

    </wsdl:port>

  </wsdl:service>

</wsdl:definitions>'''

    SOAP_RESPONSE = '''\

<?xml version="1.0" encoding="UTF-8"?>

<soap:Envelope

  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

  xmlns:xsd="http://www.w3.org/2001/XMLSchema"

  xmlns:soap="http://www.w3.org/2003/05/soap-envelope"

  xmlns:ser="http://www.herongyang.com/Service/">

  <soap:Header/>

  <soap:Body>

     <sayHelloResponse xmlns="urn:examples:helloservice">

       <greeting>Hello John!</greeting>

     </sayHelloResponse>

  </soap:Body>

</soap:Envelope>'''

    INPUT_SCHEMA = {

        'type': 'object',

        'properties': {

            'firstName': {'type': 'array', 'items': {'type': 'string'}},

            'lastName': {'type': 'string'},

        },

        'required': ['firstName', 'lastName'],

    }

    INPUT_DATA = {

        'firstName/0': 'John',

        'firstName/1': 'Bill',

        'lastName': 'Doe',

    }

@pytest.fixture(params=[SOAP11, SOAP12])

def soap(request):

    p = request.param()

    with utils.mock_url(p.WSDL_URL, response=p.WSDL_CONTENT):

        with utils.mock_url(p.ENDPOINT_URL, response=p.SOAP_RESPONSE) as mock:

            p.endpoint_mock = mock

            yield p

@pytest.fixture

def connector(db, soap):

    return utils.setup_access_rights(

        SOAPConnector.objects.create(

            slug='test', wsdl_url=soap.WSDL_URL, zeep_strict=True, zeep_xsd_ignore_sequence_order=False

        )

def test_schemas(connector, soap):

    assert list(connector.operations_and_schemas) == [('sayHello', soap.INPUT_SCHEMA, soap.OUTPUT_SCHEMA)]

def test_say_hello_method_validation_error(connector, app):

    resp = app.get('/soap/test/method/sayHello/', status=500)

    assert dict(resp.json, err_desc=None) == {

        'err': 1,

        'err_class': 'zeep.exceptions.ValidationError',

        'err_desc': None,

        'data': None,

    }

def test_say_hello_method_ok_get(connector, app, caplog, soap):

    resp = app.get('/soap/test/method/sayHello/?' + urllib.parse.urlencode(soap.INPUT_DATA))

    assert '>John<' in soap.endpoint_mock.handlers[0].call['requests'][-1].body.decode()

    assert '>Bill<' in soap.endpoint_mock.handlers[0].call['requests'][-1].body.decode()

    assert '>Doe<' in soap.endpoint_mock.handlers[0].call['requests'][-1].body.decode()

    assert resp.json == {'data': 'Hello John!', 'err': 0}

def test_say_hello_method_ok_post_json(connector, app, caplog, soap):

    resp = app.post_json('/soap/test/method/sayHello/', params=soap.INPUT_DATA)

    assert '>John<' in soap.endpoint_mock.handlers[0].call['requests'][-1].body.decode()

    assert '>Bill<' in soap.endpoint_mock.handlers[0].call['requests'][-1].body.decode()

    assert '>Doe<' in soap.endpoint_mock.handlers[0].call['requests'][-1].body.decode()

    assert resp.json == {'data': 'Hello John!', 'err': 0}

# Copyright (C) 2021  Entr'ouvert

#

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

import mock

import pytest

import requests

from django.utils.encoding import force_bytes

from zeep import Settings

from zeep.exceptions import TransportError, XMLParseError

from zeep.plugins import Plugin

from passerelle.utils.soap import SOAPClient

WSDL = 'tests/data/soap.wsdl'

class FooPlugin(Plugin):

    pass

class BarPlugin(Plugin):

    pass

class SOAPResource(object):

    def __init__(self):

        self.requests = requests.Session()

        self.wsdl_url = WSDL

def test_soap_client():

    soap_resource = SOAPResource()

    plugins = [FooPlugin, BarPlugin]

    client = SOAPClient(soap_resource, plugins=plugins)

    assert client.wsdl.location.endswith(WSDL)

    assert client.transport.session == soap_resource.requests

    assert client.transport.cache

    assert client.plugins == plugins

@mock.patch('requests.sessions.Session.post')

def test_disable_strict_mode(mocked_post):

    response = requests.Response()

    response.status_code = 200

    response._content = force_bytes(

        '''<?xml version='1.0' encoding='utf-8'?>

<soap-env:Envelope xmlns:soap-env="http://schemas.xmlsoap.org/soap/envelope/">

  <soap-env:Body>

    <ns0:TradePrice xmlns:ns0="http://example.com/stockquote.xsd">

      <price>4.20</price>

    </ns0:TradePrice>

  </soap-env:Body>

</soap-env:Envelope>'''

    )

    mocked_post.return_value = response

    soap_resource = SOAPResource()

    client = SOAPClient(soap_resource)

    match = "Unexpected element %s, expected %s" % (repr(u'price'), repr(u'skipMe'))

    with pytest.raises(XMLParseError, match=match):

        client.service.GetLastTradePrice(tickerSymbol='banana')

    client = SOAPClient(soap_resource, settings=Settings(strict=False))

    result = client.service.GetLastTradePrice(tickerSymbol='banana')

    assert len(result) == 2

    assert result['skipMe'] is None

    assert result['price'] == 4.2

@mock.patch('requests.sessions.Session.post')

def test_remove_first_bytes_for_xml(mocked_post):

    response = requests.Response()

    response.status_code = 200

    response._content = force_bytes(

        '''blabla \n<?xml version='1.0' encoding='utf-8'?>

<soap-env:Envelope xmlns:soap-env="http://schemas.xmlsoap.org/soap/envelope/">

  <soap-env:Body>

    <ns0:TradePrice xmlns:ns0="http://example.com/stockquote.xsd">

      <skipMe>1.2</skipMe>

      <price>4.20</price>

    </ns0:TradePrice>

  </soap-env:Body>

</soap-env:Envelope>\n bloublou'''

    )

    mocked_post.return_value = response

    soap_resource = SOAPResource()

    client = SOAPClient(soap_resource)

    with pytest.raises(TransportError):

        client.service.GetLastTradePrice(tickerSymbol='banana')

    client = SOAPClient(soap_resource, transport_kwargs={'remove_first_bytes_for_xml': True})

    result = client.service.GetLastTradePrice(tickerSymbol='banana')

    assert len(result) == 2

    assert result['skipMe'] == 1.2

    assert result['price'] == 4.2