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 datetime import datetime

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)

        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 = {

            'Accept': 'application/json; charset=utf-8',

            'token_info_caller': base64.b64encode(msg),

        }

        if method == 'get':

            response = self.requests.get(url, headers=headers, params=params)

        elif method == 'post':

            response = self.requests.post(url, headers=headers, json=json)

        elif method == 'delete':

            response = self.requests.delete(url, headers=headers)

            if response.status_code == 304:  # handle strange 304 delete response

                raise APIError('Appointment not found')

        else:

            raise APIError('Method Not Allowed', http_status=405)

        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'),

        name='create-appointment',

        perm='can_access',

        methods=['post'],

        description=_('Create appointment'),

        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'),

        name='get-appointment',

        perm='can_access',

        methods=['get'],

        description=_('Get appointment'),

        parameters={

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

        },

    )

    def get_appointment(self, request, id):

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

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

    @endpoint(

        display_category=_('Appointment'),

        name='delete-appointment',

        perm='can_access',

        methods=['post'],

        description=_('Delete appointment'),

        parameters={

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

        },

    )

    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" : "943A98",

   "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")

def test_pre_request(connector):

    @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 request.headers['token_info_caller'][:42] == b'f3G6sjRZETBam6vcdrAxmvJQTX5hh6OjZ8XlUO6SMo'

        return httmock.response(200)

    with httmock.HTTMock(esirius_mock):

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

@pytest.mark.parametrize(

    'method, status_code',

    [

        ('get', 200),

        ('post', 200),

        ('delete', 200),

        ('put', 405),

    ],

)

def test_request_method(connector, method, status_code):

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

    def esirius_mock(url, request):

        return httmock.response(status_code, '{}')

    if status_code == 200:

        with httmock.HTTMock(esirius_mock):

            resp = connector.request('an/uri/', method=method)

            assert resp.status_code == 200

    else:

        with pytest.raises(APIError, match='Method Not Allowed') as exc:

            with httmock.HTTMock(esirius_mock):

                connector.request('an/uri/', method=method)

        assert exc.value.err

        assert exc.value.http_status == status_code

@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, content, is_up',

    [

        (200, 'wathever', True),

        (500, '{"message": "help"}', False),

    ],

)

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

    @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/', method='GET')

    def esirius_mock(url, request):

        return httmock.response(200, GET_APPOINTMENT_RESPONSE)

    with httmock.HTTMock(esirius_mock):

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

    assert not resp.json['err']

    assert resp.json['data']['codeRDV']

    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/', 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 + '?id=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/', method='DELETE')

    def esirius_mock(url, request):

        return httmock.response(200, b'')

    with httmock.HTTMock(esirius_mock):

        resp = app.post_json(endpoint + '?id=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/', method='DELETE')

    def esirius_mock(url, request):

        return httmock.response(304, b'')

    with httmock.HTTMock(esirius_mock):

        resp = app.post_json(endpoint + '?id=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!'