Produits Entr'ouvert » Passerelle

# -*- coding: utf-8 -*-

# Generated by Django 1.11.18 on 2021-03-01 14:52

from __future__ import unicode_literals

from django.db import migrations, models

import passerelle.apps.esirius.models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0029_auto_20210202_1627'),

    ]

    operations = [

        migrations.CreateModel(

            name='ESirius',

            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(default=True, verbose_name='TLS verify certificates')),

                (

                    'http_proxy',

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

                ),

                (

                    'secret_id',

                    models.CharField(blank=True, max_length=128, verbose_name='Application identifier'),

                ),

                (

                    'secret_key',

                    passerelle.apps.esirius.models.DESKeyModel(

                        blank=True, max_length=128, verbose_name='Secret Key'

                    ),

                ),

                (

                    'base_url',

                    models.CharField(

                        help_text='example: https://HOST/ePlanning/webservices/api/',

                        max_length=256,

                        verbose_name='ePlanning webservices URL',

                    ),

                ),

                (

                    'users',

                    models.ManyToManyField(

                        blank=True, related_name='_esirius_users_+', related_query_name='+', to='base.ApiUser'

                    ),

                ),

            ],

            options={

                'verbose_name': 'eSirius',

            },

        ),

    ]

# passerelle - uniform access to multiple data sources and services

# 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 base64

from time import time

from urllib.parse import urljoin

from Cryptodome.Cipher import DES

from Cryptodome.Util.Padding import pad

from django.db import models

from django.core.exceptions import ValidationError

from django.utils.encoding import force_bytes

from django.utils.translation import ugettext_lazy as _

from passerelle.base.models import BaseResource, HTTPResource

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

CREATE_APPOINTMENT_SCHEMA = {

    '$schema': 'http://json-schema.org/draft-04/schema#',

    "type": "object",

    'properties': {

        'idSys': {'type': 'string', 'pattern': '^[0-9]*$'},

        'CodeRDV': {'type': 'string'},

        'beginDate': {'type': 'string', 'pattern': '^[0-9]{4}-[0-9]{2}-[0-9]{2}$'},

        'beginTime': {'type': 'string', 'pattern': '^[0-9]{2}:[0-9]{2}$'},

        'endDate': {'type': 'string', 'pattern': '^[0-9]{4}-[0-9]{2}-[0-9]{2}$'},

        'endTime': {'type': 'string', 'pattern': '^[0-9]{2}:[0-9]{2}$'},

        'comment': {'type': 'string'},

        'isoLanguage': {'description': 'ex: fr', 'type': 'string'},

        'needsConfirmation': {'description': 'boolean expected', 'type': 'string'},

        'rdvChannel': {'description': 'ex: EAPP0', 'type': 'string'},

        'receptionChannel': {'type': 'string'},

        'owner': {'type': 'object', 'properties': {'key': {'type': 'string'}, 'value': {'type': 'string'}}},

        'user': {

            'type': 'object',

            'properties': {

                'idSys': {'type': 'string', 'pattern': '^[0-9]*$'},

                'personalIdentity': {'type': 'string'},

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

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

                'civility': {'type': 'string'},

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

                'birthday': {'type': 'string'},

                'email': {'type': 'string'},

                'fixPhone': {'type': 'string'},

                'phone': {'type': 'string'},

                'address': {

                    'type': 'object',

                    'properties': {

                        'line1': {'type': 'string'},

                        'line2': {'type': 'string'},

                        'zipCode': {'type': 'string'},

                        'city': {'type': 'string'},

                        'country': {'type': 'string'},

                    },

                },

            },

        },

        'serviceId': {'type': 'string'},

        'siteCode': {'type': 'string'},

        "resources": {

            'type': 'object',

            'properties': {

                'id': {'type': 'string', 'pattern': '^[0-9]*$'},

                'key': {'type': 'string'},

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

                'name': {'type': 'string'},

                'station': {

                    'type': 'object',

                    'properties': {

                        'id': {'type': 'string', 'pattern': '^[0-9]*$'},

                        'key': {'type': 'string'},

                        'name': {'type': 'string'},

                    },

                },

            },

        },

        'motives': {

            "type": "array",

            "items": {

                'type': 'object',

                'properties': {

                    'id': {'type': 'string', 'pattern': '^[0-9]*$'},

                    'name': {'type': 'string'},

                    'shortName': {'type': 'string'},

                    'processingTime': {'type': 'string', 'pattern': '^[0-9]*$'},

                    'externalModuleAccess': {'type': 'string', 'pattern': '^[0-9]*$'},

                    'quantity': {'type': 'string', 'pattern': '^[0-9]*$'},

                    'usePremotiveQuantity': {'description': 'boolean expected', 'type': 'string'},

                },

            },

        },

    },

    'unflatten': True,

}

class DESKeyModel(models.CharField):

    def clean(self, value, model_instance):

        if len(value) < 8:

            raise ValidationError(_('DES key must be 8 bytes long (longer keys are truncated)'))

        return super().clean(value, model_instance)

class ESirius(BaseResource, HTTPResource):

    secret_id = models.CharField(max_length=128, verbose_name=_('Application identifier'), blank=True)

    secret_key = DESKeyModel(max_length=128, verbose_name=_('Secret Key'), blank=True)

    base_url = models.CharField(

        max_length=256,

        blank=False,

        verbose_name=_('ePlanning webservices URL'),

        help_text=_('example: https://HOST/ePlanning/webservices/api/'),

    )

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = _('eSirius')

    def request(self, uri, method='GET', params=None, json=None):

        url = urljoin(self.base_url, uri)

        headers = {'Accept': 'application/json; charset=utf-8'}

        if self.secret_key:

            des_key = pad(force_bytes(self.secret_key), 8)[:8]

            cipher = DES.new(des_key, DES.MODE_ECB)

            epoch = int(time() * 1000)

            plaintext = '{"caller":"%s","createInfo":%i}' % (self.secret_id, epoch)

            msg = cipher.encrypt(pad(force_bytes(plaintext), 8))

            headers['token_info_caller'] = base64.b64encode(msg)

        response = self.requests.request(method=method, url=url, headers=headers, params=params, json=json)

        # handle strange 304 delete response

        if method == 'DELETE' and response.status_code == 304:

            raise APIError('Appointment not found')

        # handle 500 generated by wrong token as a succesfull ping response

        if uri == 'sites/' and response.status_code == 500:

            response.status_code = 200

        if response.status_code != 200:

            try:

                json_content = response.json()

            except ValueError:

                json_content = None

            raise APIError(

                'error status:%s %r, content:%r'

                % (response.status_code, response.reason, response.text[:1024]),

                data={'status_code': response.status_code, 'json_content': json_content},

            )

        return response

    def check_status(self):

        """

        Raise an exception if something goes wrong.

        """

        self.request('sites/', method='GET')

    @endpoint(

        display_category=_('Appointment'),

        description=_('Create appointment'),

        name='create-appointment',

        perm='can_access',

        methods=['post'],

        post={'request_body': {'schema': {'application/json': CREATE_APPOINTMENT_SCHEMA}}},

    )

    def create_appointment(self, request, post_data):

        # address dict is required

        if not post_data.get('user'):

            post_data['user'] = {}

        if not post_data['user'].get('address'):

            post_data['user']['address'] = {}

        response = self.request('appointments/', method='POST', json=post_data)

        return {'data': {'id': response.text}}

    @endpoint(

        display_category=_('Appointment'),

        description=_('Get appointment'),

        name='get-appointment',

        perm='can_access',

        methods=['get'],

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

        example_pattern='{id}/',

        parameters={

            'id': {

                'description': _('Appointment id returned by create-appointment endpoint'),

                'example_value': '94PEP4',

            },

        },

    )

    def get_appointment(self, request, id):

        response = self.request('appointments/%s/' % id, method='GET')

        return {'data': response.json()}

    @endpoint(

        display_category=_('Appointment'),

        description=_('Delete appointment'),

        name='delete-appointment',

        perm='can_access',

        methods=['delete'],

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

        example_pattern='{id}/',

        parameters={

            'id': {

                'description': _('Appointment id returned by create-appointment endpoint'),

                'example_value': '94PEP4',

            },

        },

    )

    def delete_appointment(self, request, id):

        response = self.request('appointments/%s/' % id, method='DELETE')

        return {'data': None}

    'passerelle.apps.esirius',

# passerelle - uniform access to multiple data sources and services

# 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 json

import httmock

import pytest

from passerelle.apps.esirius.models import ESirius

from passerelle.utils.jsonresponse import APIError

from test_manager import login

import utils

CREATE_APPOINTMENT_PAYLOAD = {

    'beginDate': '2021-02-24',

    'beginTime': '16:40',

    'endDate': '2021-02-24',

    'endTime': '17:00',

    'comment': 'commentaire',

    'isoLanguage': 'fr',

    'needsConfirmation': 'False',

    'rdvChannel': 'WEBSERVICES',

    'receptionChannel': 'WS',

    'serviceId': '9',

    'siteCode': 'site1',

    'resources': {

        'id': '1',

        'key': '17',

        'type': 'STATION',

    },

}

GET_APPOINTMENT_RESPONSE = '''

{

   "beginDate" : "2021-02-26",

   "beginTime" : "16:40",

   "codeRDV" : "94PEP4",

   "comment" : "coucou",

   "endDate" : "2021-02-26",

   "endTime" : "17:00",

   "idSys" : 108840,

   "isoLanguage" : "fr",

   "motives" : [],

   "needsConfirmation" : false,

   "rdvChannel" : "EAPP0",

   "receptionChannel" : "WS",

   "resources" : {

      "id" : 29,

      "key" : "46",

      "name" : "C1",

      "type" : "STATION"

   },

   "serviceId" : "39",

   "siteCode" : "site1",

   "siteIdSys" : 5,

   "user" : {

      "additionalPersonalIdentity" : [],

      "address" : {},

      "civility" : "",

      "idSys" : "95897"

   }

}

'''

@pytest.fixture

def connector(db):

    return utils.setup_access_rights(

        ESirius.objects.create(

            slug='test', secret_id='xxx', secret_key='yyy', base_url='https://dummy-server.org'

        )

    )

def get_endpoint(name):

    return utils.generic_endpoint_url('esirius', name)

@pytest.mark.freeze_time('2021-01-26 15:13:6.880')  # epoch + 1611673986.88 s

def test_token(connector):

    connector.secret_id = 'eAppointment'

    connector.secret_key = 'ES2I Info Caller Http Encryption Key'

    connector.save()

    @httmock.all_requests

    def esirius_mock(url, request):

        assert (

            request.headers['token_info_caller']

            == b'yM4zYAxT67Qvjd20riG3j0eu0t0Ku+HLlttj17Gul7zkruFaXX1J8BJ6sV2Ldgw40axfWh+ESAY='

        )

        return httmock.response(200)

    with httmock.HTTMock(esirius_mock):

        connector.request('an/uri/', method='get', params="somes")

@pytest.mark.parametrize('secret_key', ['xxx', ''])

def test_pre_request(connector, secret_key):

    @httmock.urlmatch(netloc='dummy-server.org', path='/an/uri/', method='GET')

    def esirius_mock(url, request):

        assert request.headers['Accept'] == 'application/json; charset=utf-8'

        assert bool(request.headers.get('token_info_caller')) == bool(secret_key)

        if secret_key:

            assert request.headers['token_info_caller'][:42] == b'f3G6sjRZETBam6vcdrAxmvJQTX5hh6OjZ8XlUO6SMo'

        return httmock.response(200)

    connector.secret_key = secret_key

    connector.save()

    with httmock.HTTMock(esirius_mock):

        connector.request('an/uri/', method='get', params="somes")

@pytest.mark.parametrize(

    'status_code, content, a_dict',

    [

        (400, '{"message": "help"}', {'message': 'help'}),

        (500, 'not json', None),

    ],

)

def test_post_request(connector, status_code, content, a_dict):

    @httmock.urlmatch(netloc='dummy-server.org', path='/an/uri/', method='GET')

    def esirius_mock(url, request):

        return httmock.response(status_code, content)

    with pytest.raises(APIError) as exc:

        with httmock.HTTMock(esirius_mock):

            connector.request('an/uri/', params="somes")

    assert exc.value.err

    assert exc.value.data['status_code'] == status_code

    assert exc.value.data['json_content'] == a_dict

@pytest.mark.parametrize('status_code, is_up', [(200, True), (500, True), (503, False)])

@pytest.mark.parametrize('content', ['wathever', '{"message": "help"}'])

def test_check_status(app, connector, status_code, is_up, content):

    @httmock.all_requests

    def esirius_mock(url, request):

        return httmock.response(status_code, content)

    if is_up:

        with httmock.HTTMock(esirius_mock):

            connector.check_status()

    else:

        with pytest.raises(APIError):

            with httmock.HTTMock(esirius_mock):

                connector.check_status()

def test_create_appointment(app, connector):

    endpoint = get_endpoint('create-appointment')

    @httmock.urlmatch(netloc='dummy-server.org', path='/appointments/', method='POST')

    def esirius_mock(url, request):

        return httmock.response(200, b'94PEP4')

    with httmock.HTTMock(esirius_mock):

        resp = app.post_json(endpoint, params=CREATE_APPOINTMENT_PAYLOAD)

    assert not resp.json['err']

    assert resp.json['data'] == {'id': '94PEP4'}

def test_create_appointment_error_404(app, connector):

    endpoint = get_endpoint('create-appointment')

    # payload not providing or providing an unconfigured serviceId

    payload = CREATE_APPOINTMENT_PAYLOAD

    del payload['serviceId']

    @httmock.urlmatch(netloc='dummy-server.org', path='/appointments/', method='POST')

    def esirius_mock(url, request):

        return httmock.response(

            404,

            {

                'code': 'Not Found',

                'type': 'com.es2i.planning.api.exception.NoService4RDVException',

                'message': "Le rendez-vous {0} n'a pas créé",

            },

        )

    with httmock.HTTMock(esirius_mock):

        resp = app.post_json(endpoint, params=payload)

    assert resp.json['err']

    assert resp.json['data']['status_code'] == 404

    assert resp.json['data']['json_content'] == {

        'code': 'Not Found',

        'type': 'com.es2i.planning.api.exception.NoService4RDVException',

        'message': "Le rendez-vous {0} n'a pas créé",

    }

def test_create_appointment_error_500(app, connector):

    endpoint = get_endpoint('create-appointment')

    # payload not providing beginTime

    payload = {'beginDate': '2021-02-23'}

    @httmock.urlmatch(netloc='dummy-server.org', path='/appointments/', method='POST')

    def esirius_mock(url, request):

        return httmock.response(500, 'java stack')

    with httmock.HTTMock(esirius_mock):

        resp = app.post_json(endpoint, params=payload)

    assert resp.json['err']

    assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'

    assert resp.json['err_desc'] == "error status:500 None, content:'java stack'"

    assert resp.json['data']['status_code'] == 500

    assert resp.json['data']['json_content'] is None

def test_get_appointment(app, connector):

    endpoint = get_endpoint('get-appointment')

    @httmock.urlmatch(netloc='dummy-server.org', path='/appointments/94PEP4/', method='GET')

    def esirius_mock(url, request):

        return httmock.response(200, GET_APPOINTMENT_RESPONSE)

    with httmock.HTTMock(esirius_mock):

        resp = app.get(endpoint + '/94PEP4/')

    assert not resp.json['err']

    assert resp.json['data']['codeRDV'] == '94PEP4'

    assert resp.json['data'] == json.loads(GET_APPOINTMENT_RESPONSE)

def test_get_appointment_error(app, connector):

    endpoint = get_endpoint('get-appointment')

    @httmock.urlmatch(netloc='dummy-server.org', path='/appointments/94PEP4/', method='GET')

    def esirius_mock(url, request):

        return httmock.response(404, '{"code":"Not Found","message":"Le rendez-vous {0} n\'existe pas"}')

    with httmock.HTTMock(esirius_mock):

        resp = app.get(endpoint + '/94PEP4/')

    assert resp.json['err']

    assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'

    assert resp.json['data']['status_code'] == 404

    assert resp.json['data']['json_content'] == {

        "code": "Not Found",

        "message": "Le rendez-vous {0} n'existe pas",

    }

def test_delete_appointment(app, connector):

    endpoint = get_endpoint('delete-appointment')

    @httmock.urlmatch(netloc='dummy-server.org', path='/appointments/94PEP4/', method='DELETE')

    def esirius_mock(url, request):

        return httmock.response(200, b'')

    with httmock.HTTMock(esirius_mock):

        resp = app.delete(endpoint + '/94PEP4/')

    assert not resp.json['err']

    assert resp.json['data'] == None

def test_delete_appointment_error(app, connector):

    endpoint = get_endpoint('delete-appointment')

    @httmock.urlmatch(netloc='dummy-server.org', path='/appointments/94PEP4/', method='DELETE')

    def esirius_mock(url, request):

        return httmock.response(304, b'')

    with httmock.HTTMock(esirius_mock):

        resp = app.delete(endpoint + '/94PEP4/')

    assert resp.json['err']

    assert resp.json['err_desc'] == 'Appointment not found'

def test_manager(db, app, admin_user, connector):

    url = '/%s/%s/' % (connector.get_connector_slug(), connector.slug)

    connector.title = 'Test'

    connector.description = 'Test eSirius'

    connector.save()

    login(app)

    resp = app.get(url)

    resp = resp.click('edit')

    assert resp.html.find('div', {'id': 'id_secret_key_p'}).input['value'] == 'yyy'

    resp = resp.form.submit()

    assert (

        'DES key must be 8 bytes long'

        in resp.html.find('div', {'id': 'id_secret_key_p'}).find('div', {'class': 'error'}).text

    )

    resp.form['secret_key'] = '8 bytes!'

    resp = resp.form.submit()

    assert ESirius.objects.get().secret_key == '8 bytes!'

    # accept an empty key

    resp = app.get(url)

    resp = resp.click('edit')

    assert resp.html.find('div', {'id': 'id_secret_key_p'}).input['value'] == '8 bytes!'

    resp.form['secret_key'] = ''

    resp = resp.form.submit()

    assert ESirius.objects.get().secret_key == ''