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.core.exceptions import PermissionDenied

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=_('FranceConnect callback (internal use)'),

        parameters={

            'origin': {

                'description': _('HTTP Origin, needed to secure window.postMessage'),

            },

            'mode': {

                'description': _('Mode'),

            },

            'test': {

                'description': _('Use test mode (to see exchanges)'),

            },

        },

    )

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

        if test and not request.user.is_superuser:

            return HttpResponseBadRequest('Only admin can use test mode.')

        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

            try:

                template = Template(self.fc_text_template)

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

            except Exception:

                token['text'] = '<failed to render>'

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

        except fc.FranceConnectError as e:

            context['error'] = e

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

    @endpoint(

        description=_('Demo page (to check your configuration)'),

    )

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

        if not request.user.is_superuser:

            return PermissionDenied

        return TemplateResponse(

            request,

            'franceconnect/demo.html',

            context={'origin': request.build_absolute_uri('/'), 'resource': self},

        )

    @endpoint(

        description=_('Data source'),

    )

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

        if id:

            return {

                'data': [

                    dict(self.retrieve(id), id=id),

                ]

            }

        url = request.build_absolute_uri(

            reverse(

                'generic-endpoint',

                kwargs={

                    'slug': self.slug,

                    'connector': self.get_connector_slug(),

                    'endpoint': 'init_request',

                },

            )

        )

        if mode == 'dgfip':

            url += '?mode=dgfip'

        return {

            'data': [

                {

                    'id': '',

                    'text': '',

                    'init_request_url': url,

                }

            ]

        }

    def store(self, data):

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

        cache.set(ref, data)

        return ref

    def retrieve(self, ref):

        return cache.get(ref)

<html>

    <head>

    </head>

    <body>

        {% if test %}<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 %}

        </ul>

        {% endif %}

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

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

{% load i18n %}

{% block breadcrumb %}

{{ block.super }}

<a href="{{ resource.get_absolute_url }}">{{ resource.title }}</a>

<a href="#">{% trans "Demo view" %}</a>

{% endblock %}

{% block appbar %}

{% endblock %}

{% block content %}

<h1>{% trans "Demo view" %}</h1>

<p>

  <button id="start">{% trans "Get FranceConnect data" %}</button>

</p>

<p>

  <label for="dgfip">{% trans "Get DGFIP data" %}</label>

  <input type="checkbox" id="dgfip"/>

</p>

        <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; };

                var url = 'init_request?test=1&origin={{ origin }}';

                var checkbox = document.getElementById('dgfip');

                if (checkbox && checkbox.checked) {

                    url += '&mode=dgfip';

                };

                popupCenter(url, 'FranceConnect', 1000, 670);

            });

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

                var data = event.originalEvent.data;

                var origin = event.originalEvent.origin;

                if (origin != window.location.origin) {

                    return;

                }

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

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

                    $('#user-info-preview').text(JSON.stringify(result, null, 2));

                });

                popup.close();

            });

        </script>

        <div id="user-info" style="display: none; position: relative;">

            <h3>{% trans "Data-source data" %}</h3>

            <pre id="user-info-preview" style="background: white; border: 0.2ex solid black; color: black; padding: 1em; width: calc(100% - 5em);"></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 %}

    'passerelle.apps.franceconnect',