Produits Entr'ouvert » Passerelle

# 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

import json

import urllib.parse

import uuid

import requests

from django.utils.translation import ugettext_lazy as _

class FranceConnectError(Exception):

    def __init__(self, message, **kwargs):

        self.data = tuple(kwargs.items())

        super().__init__(message)

class Test:

    slug = 'test'

    name = _('Testing')

    authorize_url = 'https://fcp.integ01.dev-franceconnect.fr/api/v1/authorize'

    token_endpoint_url = 'https://fcp.integ01.dev-franceconnect.fr/api/v1/token'

    user_info_endpoint_url = 'https://fcp.integ01.dev-franceconnect.fr/api/v1/userinfo'

    logout_url = 'https://fcp.integ01.dev-franceconnect.fr/api/v1/logout'

class Prod:

    slug = 'prod'

    name = _('Production')

    authorize_url = 'https://app.franceconnect.gouv.fr/api/v1/authorize'

    token_endpoint_url = 'https://app.franceconnect.gouv.fr/api/v1/token'

    user_info_endpoint_url = 'https://app.franceconnect.gouv.fr/api/v1/userinfo'

    logout_url = 'https://app.franceconnect.gouv.fr/api/v1/logout'

PLATFORMS = [Test, Prod]

PLATFORMS_BY_SLUG = {platform.slug: platform for platform in PLATFORMS}

def base64url_decode(input):

    rem = len(input) % 4

    if rem > 0:

        input += b'=' * (4 - rem)

    return base64.urlsafe_b64decode(input)

class FranceConnect:

    def __init__(self, session, logger):

        self.session = session

        self.logger = logger

        self.items = []

        self.correlation_id = str(uuid.uuid4())

    def authorization_request(self, platform, client_id, scopes, redirect_uri, acr_values='eidas1'):

        '''Launch an authorization request to FranceConnect'''

        qs = urllib.parse.urlencode(

            {

                'response_type': 'code',

                'client_id': client_id,

                'redirect_uri': redirect_uri,

                'scope': 'openid ' + scopes,

                'state': str(uuid.uuid4()),

                'nonce': str(uuid.uuid4()),

                'acr_values': acr_values,

            }

        )

        return '%s?%s' % (platform.authorize_url, qs)

    def handle_authorization_response(

        self, platform, client_id, client_secret, redirect_uri, code, error, error_description

    ):

        if error:

            raise FranceConnectError(

                'No authorization code', error=error, error_description=error_description

            )

        data = {

            'grant_type': 'authorization_code',

            'redirect_uri': redirect_uri,

            'client_id': client_id,

            'client_secret': client_secret,

            'code': code,

        }

        response_content = self.request('token endpoint', 'POST', platform.token_endpoint_url, data=data)

        try:

            self.add('fc_token_endpoint_response', response_content)

            self.add('fc_access_token', response_content['access_token'])

            self.add('fc_id_token', response_content['id_token'])

            header, payload, signature = self.fc_id_token.split('.')

            self.add('fc_id_token_payload', json.loads(base64url_decode(payload.encode())))

        except Exception as e:

            raise FranceConnectError('Error in token endpoint response', sub_exception=e)

        fc_user_info = self.request(

            'user_info endpoint',

            'GET',

            platform.user_info_endpoint_url,

            headers={'Authorization': 'Bearer %s' % self.fc_access_token},

        )

        self.add('fc_user_info', fc_user_info)

    def request_dgfip_access_token(self, dgfip_username, dgfip_password, scope=None):

        data = {

            'grant_type': 'client_credentials',

        }

        if scope:

            data['scope'] = scope

        dgfip_response = self.request(

            'dgfip token endpoint',

            'POST',

            'https://gwfc.impots.gouv.fr/token',

            data=data,

            auth=(dgfip_username, dgfip_password),

        )

        self.add('dgfip_token_endpoint_response', dgfip_response)

        try:

            dgfip_access_token = dgfip_response['access_token']

        except (TypeError, KeyError) as e:

            raise FranceConnectError('dgfip token endpoint error %s' % e, response=dgfip_response)

        self.add('dgfip_access_token', dgfip_access_token)

    def request_dgfip_ir(self, annrev, id_teleservice=None):

        headers = {

            'Authorization': 'Bearer %s' % self.dgfip_access_token,

            'X-FranceConnect-OAuth': self.fc_access_token,

            'X-Correlation-ID': str(uuid.uuid4()),

            'Accept': 'application/prs.dgfip.part.situations.ir.assiettes.v1+json',

        }

        if id_teleservice:

            headers['ID_Teleservice'] = id_teleservice

        try:

            dgfip_ressource_ir_response = self.request(

                'ressource IR endpoint',

                'GET',

                'https://gwfc.impots.gouv.fr/impotparticulier/1.0/situations/ir/assiettes/annrev/%s' % annrev,

                headers=headers,

            )

        except FranceConnectError as e:

            dgfip_ressource_ir_response = {'error_desc': str(e), 'error': e.data}

        # accumulate data

        try:

            data = self.dgfip_ressource_ir_response

        except AttributeError:

            data = {}

        data[annrev] = dgfip_ressource_ir_response

        self.add('dgfip_ressource_ir_response', data)

    def __getattr__(self, name):

        try:

            return dict(self.items)[name]

        except KeyError:

            raise AttributeError(name)

    def add(self, key, value):

        self.items.append((key, value))

    def request(self, label, method, url, *args, **kwargs):

        self.logger.debug('request %s %s args:%s kwargs:%s', label, method, args, kwargs)

        self.add(label.replace(' ', '_') + '_request', [method, url, args, kwargs])

        try:

            response = getattr(self.session, method.lower())(url, *args, **kwargs)

            try:

                response_content = response.json()

            except ValueError:

                response_content = response.text[:1024]

                response.raise_for_status()

                raise

            else:

                response.raise_for_status()

        except requests.HTTPError as e:

            raise FranceConnectError('%s error %s' % (label, e), response=response_content)

        except requests.RequestException as e:

            raise FranceConnectError('%s error %s' % (label, e))

        except ValueError as e:

            raise FranceConnectError('%s error %s' % (label, e), response=response_content)

        return response_content

# Generated by Django 2.2.19 on 2021-05-17 11:43

from django.db import migrations, models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0029_auto_20210202_1627'),

    ]

    operations = [

        migrations.CreateModel(

            name='Resource',

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

                (

                    'fc_platform_slug',

                    models.CharField(

                        choices=[('test', 'Testing'), ('prod', 'Production')],

                        max_length=4,

                        verbose_name='FranceConnect platform',

                    ),

                ),

                ('fc_client_id', models.CharField(max_length=64, verbose_name='FranceConnect client_id')),

                (

                    'fc_client_secret',

                    models.CharField(max_length=64, verbose_name='FranceConnect client_secret'),

                ),

                (

                    'fc_scopes',

                    models.TextField(default='identite_pivot', verbose_name='FranceConnect scopes'),

                ),

                (

                    'fc_text_template',

                    models.TextField(

                        default="{{ given_name }} {{ family_name }} {% if gender == 'male' %}né{% else %}née{% endif %} le {{ birthdate }} à {{ birthplace }}",

                        verbose_name='FranceConnect text template',

                    ),

                ),

                (

                    'dgfip_username',

                    models.CharField(

                        blank=True, max_length=64, null=True, verbose_name='api.impots.gouv.fr username'

                    ),

                ),

                (

                    'dgfip_password',

                    models.CharField(

                        blank=True, max_length=64, null=True, verbose_name='api.impots.gouv.fr password'

                    ),

                ),

                (

                    'dgfip_scopes',

                    models.TextField(blank=True, null=True, verbose_name='api.impots.gouv.fr scopes'),

                ),

                (

                    'dgfip_id_teleservice',

                    models.TextField(blank=True, null=True, verbose_name='api.impots.gouv.fr ID_Teleservice'),

                ),

                (

                    'users',

                    models.ManyToManyField(

                        blank=True,

                        related_name='_resource_users_+',

                        related_query_name='+',

                        to='base.ApiUser',

                    ),

                ),

            ],

            options={

                'verbose_name': 'FranceConnect',

            },

        ),

    ]

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

from django.core.cache import cache

from django.db import models

from django.http import HttpResponseBadRequest, HttpResponseRedirect

from django.template import Context, Template

from django.template.response import TemplateResponse

from django.urls import reverse

from django.utils.http import urlencode

from django.utils.timezone import now

from django.utils.translation import ugettext_lazy as _

from passerelle.base.models import BaseResource

from passerelle.utils import get_trusted_services

from passerelle.utils.api import endpoint

from passerelle.utils.origin import is_same_origin

from . import fc

# from passerelle.utils.jsonresponse import APIError

class Resource(BaseResource):

    category = _('Business Process Connectors')

    fc_platform_slug = models.CharField(

        _('FranceConnect platform'),

        max_length=4,

        choices=[(platform.slug, platform.name) for platform in fc.PLATFORMS],

    )

    fc_client_id = models.CharField(_('FranceConnect client_id'), max_length=64)

    fc_client_secret = models.CharField(_('FranceConnect client_secret'), max_length=64)

    fc_scopes = models.TextField(_('FranceConnect scopes'), default='identite_pivot')

    fc_text_template = models.TextField(

        _('FranceConnect text template'),

        default=(

            '''{{ given_name }} {{ family_name }} '''

            '''{% if gender == 'male' %}né{% else %}née{% endif %} le {{ birthdate }} '''

            '''à {{ birthplace }}'''

        ),

    )

    dgfip_username = models.CharField(_('api.impots.gouv.fr username'), max_length=64, blank=True, null=True)

    dgfip_password = models.CharField(_('api.impots.gouv.fr password'), max_length=64, blank=True, null=True)

    dgfip_scopes = models.TextField(_('api.impots.gouv.fr scopes'), blank=True, null=True)

    dgfip_id_teleservice = models.TextField(_('api.impots.gouv.fr ID_Teleservice'), blank=True, null=True)

    log_requests_errors = False

    class Meta:

        verbose_name = _('FranceConnect')

    @property

    def fc_platform(self):

        return fc.PLATFORMS_BY_SLUG[self.fc_platform_slug]

    def build_callback_url(self, request, **kwargs):

        redirect_uri = request.build_absolute_uri(

            reverse(

                'generic-endpoint',

                kwargs={'slug': self.slug, 'connector': self.get_connector_slug(), 'endpoint': 'callback'},

            )

        )

        if kwargs:

            redirect_uri += '?' + urlencode(

                {key: value for key, value in kwargs.items() if value is not None}

            )

        return redirect_uri

    def is_trusted_origin(self, request, origin):

        for service in get_trusted_services():

            if is_same_origin(origin, service['url']):

                return True

        if is_same_origin(request.build_absolute_uri(), origin):

            return True

        return False

    @endpoint(

        description=_('Init request'),

        parameters={

            'mode': {

                'description': _('What to retrieve, default to FranceConnect identity, can be "dgfip"'),

            },

            'origin': {

                'description': _('Origin for returning results through window.postMessage'),

            },

            'test': {

                'description': _('If set to one, activate the test callback view.'),

            },

        },

    )

    def init_request(self, request, origin, mode=None, test=None):

        if not request.user.is_superuser and not self.is_trusted_origin(request, origin):

            return HttpResponseBadRequest('Missing or invalid origin')

        redirect_uri = self.build_callback_url(request, origin=origin, mode=mode, test=test)

        franceconnect = fc.FranceConnect(session=self.requests, logger=self.logger)

        return HttpResponseRedirect(

            franceconnect.authorization_request(

                platform=self.fc_platform,

                client_id=self.fc_client_id,

                scopes=self.fc_scopes,

                redirect_uri=redirect_uri,

            )

        )

    @endpoint(

        description=_('Test page'),

        parameters={

            'mode': {

                'description': _('Mode'),

            }

        },

    )

    def callback(self, request, origin, mode=None, test=None, **kwargs):

        if not request.user.is_superuser and not self.is_trusted_origin(request, origin):

            return HttpResponseBadRequest('Missing or invalid origin')

        franceconnect = fc.FranceConnect(session=self.requests, logger=self.logger)

        redirect_uri = self.build_callback_url(request, origin=origin, mode=mode, test=test)

        context = {

            'origin': origin,

            'franceconnect': franceconnect,

            'redirect_uri': redirect_uri,

            'test': test,

        }

        try:

            franceconnect.handle_authorization_response(

                platform=self.fc_platform,

                client_id=self.fc_client_id,

                client_secret=self.fc_client_secret,

                redirect_uri=redirect_uri,

                code=request.GET.get('code'),

                error=request.GET.get('error'),

                error_description=request.GET.get('error_description'),

            )

            token = {'franceconnect': franceconnect.fc_user_info}

            if mode == 'dgfip':

                franceconnect.request_dgfip_access_token(

                    self.dgfip_username, self.dgfip_password, scope=self.dgfip_scopes

                )

                current_year = now().year

                for year in range(current_year - 3, current_year):

                    franceconnect.request_dgfip_ir(str(year), id_teleservice=self.dgfip_id_teleservice)

                token['dgfip_ir'] = franceconnect.dgfip_ressource_ir_response

            context['data'] = {'id': self.store(token)}

            try:

                template = Template(self.fc_text_template)

                context['data']['text'] = template.render(Context(franceconnect.fc_user_info))

            except Exception:

                pass

        except fc.FranceConnectError as e:

            context['error'] = e

        return TemplateResponse(request, 'franceconnect/test.html', context=context)

    @endpoint(

        description=_('Demo page'),

    )

    def demo(self, request, **kwargs):

        return TemplateResponse(

            request, 'franceconnect/demo.html', context={'origin': request.build_absolute_uri('/')}

        )

    @endpoint(

        description=_('Data source'),

    )

    def data_source(self, request, id=None, **kwargs):

        if id:

            return {

                'data': dict(self.retrieve(id), id=id),

            }

        return {

            'data': [

                {

                    'id': '',

                    'text': '',

                    'init_request_url': request.build_absolute_uri(

                        reverse(

                            'generic-endpoint',

                            kwargs={

                                'slug': self.slug,

                                'connector': self.get_connector_slug(),

                                'endpoint': 'init_request',

                            },

                        )

                    ),

                }

            ]

        }

    def store(self, data):

        ref = str(uuid.uuid4().hex)

        cache.set(ref, data)

        return ref

    def retrieve(self, ref):

        return cache.get(ref)

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

{% block content %}

<button id="start">Start</button>

        <script>

            var popup = null;

const popupCenter = function(url, title, w, h) {

    // Fixes dual-screen position                             Most browsers      Firefox

    const dualScreenLeft = window.screenLeft !==  undefined ? window.screenLeft : window.screenX;

    const dualScreenTop = window.screenTop !==  undefined   ? window.screenTop  : window.screenY;

    const width = window.innerWidth ? window.innerWidth : document.documentElement.clientWidth ? document.documentElement.clientWidth : screen.width;

    const height = window.innerHeight ? window.innerHeight : document.documentElement.clientHeight ? document.documentElement.clientHeight : screen.height;

    const systemZoom = width / window.screen.availWidth;

    const left = (width - w) / 2 / systemZoom + dualScreenLeft

    const top = (height - h) / 2 / systemZoom + dualScreenTop

    popup = window.open(url, title,

      `

      scrollbars=yes,

      width=${w / systemZoom},

      height=${h / systemZoom},

      top=${top},

      left=${left}

      `

    )

    if (window.focus) popup.focus();

};

            $('#start').on('click', function () {

                $('#user-info').hide();

                if (popup) { popup.close(); popup = null; }

                popupCenter('init_request?mode=dgfip&test=1&origin={{ origin }}', 'FranceConnect', 1000, 670);

            });

            $(window).on('message', function(event) {

                var data = event.originalEvent.data;

                $.getJSON("data_source?id=" + data.id, function(result) {

                    $('#user-info').show();

                    $('#user-info-preview').text("ref: " + data.id+ "\n" + JSON.stringify(result, null, 2));

                });

                popup.close();

            });

        </script>

        <div id="user-info" style="display: none">User-Info

            <pre id="user-info-preview"></pre>

        </div>

{% endblock %}

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

{% load i18n passerelle %}

{% block description %}

{{ block.super }}

{% url "generic-endpoint" connector="franceconnect" slug=object.slug endpoint="callback"  as callback_url %}

<p>URL de callback pour FranceConnect: <a href="{{ request.scheme }}://{{ request.get_host }}{{ callback_url }}">{{ request.scheme }}://{{ request.get_host }}{{ callback_url }}</a></p>

{% endblock %}

<html>

    <head>

    </head>

    <body>

        <p><a href="?">Retry</a></p>

        <button id="continue">Continue</button>

        {{ data|json_script:"data" }}

        <p>redirect_uri: <pre>{{ redirect_uri|pprint }}</pre></p>

        <p>correlation_id: <pre>{{ franceconnect.correlation_id }}</pre></p>

        {% if error %}

            <p>{{ error }}<p>

            {% if error.data %}

                <dl>

                    {% for key, value in error.data %}

                        <dt>{{ key }}</td>

                        <dd><pre>{{ value|pprint }}</pre></dd>

                    {% endfor %}

                </dl>

            {% endif %}

        {% endif %}

        <ul>

        {% for key, value in franceconnect.items reversed %}

            <li>{{ key }}&nbsp;: <pre>{{ value|pprint }}</pre></li>

        {% endfor %}

        <script>

            (function () {

                const data = JSON.parse(document.getElementById('data').textContent);

                const continue_button = document.getElementById('continue');

                const post_message = function () {

                    window.opener.postMessage(data, "{{ origin }}");

                }

                {% if test %}continue_button.addEventListener('click', function () { post_message() });

                {% else %}post_message(){% endif %}

            })();

        </script>

    </body>

</html>

    'passerelle.apps.franceconnect',