Produits Entr'ouvert » Authentic 2

    A2_USER_PROFILE_MANAGEMENT=Setting(

        default=False,

        definition='Activate user profiles for juridical entity management.',

    ),

# Generated by Django 2.2.24 on 2022-02-15 11:39

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [

        ('custom_user', '0030_auto_20220222_1028'),

    ]

    operations = [

        migrations.AddField(

            model_name='profile',

            name='email',

            field=models.EmailField(blank=True, max_length=254, verbose_name='profile email address'),

        ),

    ]

    email = models.EmailField(blank=True, max_length=254, verbose_name=_('profile email address'))

    @property

    def PROFILE_OVERRIDE_MAPPING(self):

        return self._setting('PROFILE_OVERRIDE_MAPPING', {'email': 'email'})

            view.kwargs[lookup_url_kwarg] = uuid.UUID(bytes=decrypted.split(b'#')[0]).hex

                new_known_uuid = uuid.UUID(bytes=decrypted.split(b'#')[0]).hex

# Generated by Django 2.2.24 on 2022-02-15 09:50

import django.db.models.deletion

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [

        ('custom_user', '0030_auto_20220222_1028'),

        ('authentic2_idp_oidc', '0014_auto_20201126_1812'),

    ]

    operations = [

        migrations.AddField(

            model_name='oidccode',

            name='profile',

            field=models.ForeignKey(

                null=True,

                on_delete=django.db.models.deletion.CASCADE,

                to='custom_user.Profile',

                verbose_name='user selected profile',

            ),

        ),

    ]

from authentic2.custom_user.models import Profile

    profile = models.ForeignKey(

        to=Profile, verbose_name=_('user selected profile'), null=True, on_delete=models.CASCADE

    )

{% load i18n static %}

{% block extra-top-head %}

 {{ block.super }}

<link rel="stylesheet" type="text/css" href="{% static 'authentic2_idp_oidc/css/style.css' %}"></link>

{% endblock %}

    {% if needs_profile_validation %}

     <div id="profile-validation">

     <p>{% trans "Additionally, you may authenticate as owner of the following juridical entity management profile, which may change the aforementioned information." %}</p>

      <div class="profile" id="profile-validation-none">

        <input type="radio" id="_none" name="profile-validation" value="" checked>

        <label for="_none">{% trans "Keep my default user profile." %}</label>

      </div>

      {% for profile in user.subprofiles.all %}

        <div class="profile" id="profile-validation-{{ profile.id }}">

          <input type="radio" id="{{ profile.id }}" name="profile-validation" value="{{ profile.id }}">

          <label for="{{ profile.id }}">

          {% blocktrans with type_name=profile.profile_type.name identifier=profile.identifier %}Profile of type {{ type_name }}: {{ identifier }}{% endblocktrans %}<br />

          {% if profile.email %}

            {% blocktrans with email=profile.email %}Email address: {{ email }}{% endblocktrans %}<br />

          {% endif %}</label>

        </div>

      {% endfor %}

     </div>

    {% endif %}

def make_sub(client, user, profile=None):

        return make_pairwise_sub(client, user, profile=profile)

def make_pairwise_sub(client, user, profile=None):

        return make_pairwise_unreversible_sub(client, user, profile=profile)

        return make_pairwise_reversible_sub(client, user, profile=profile)

def make_pairwise_unreversible_sub(client, user, profile=None):

    if profile:

        sub += str(profile.id)

def make_pairwise_reversible_sub(client, user, profile=None):

    return make_pairwise_reversible_sub_from_uuid(client, user.uuid, profile=profile)

def make_pairwise_reversible_sub_from_uuid(client, user_uuid, profile=None):

    cipher_args = [

        settings.SECRET_KEY.encode('utf-8'),

        identifier,

        sector_identifier,

    ]

    if profile:

        cipher_args[1] += '#%s' % str(profile.id)

    return crypto.aes_base64url_deterministic_encrypt(*cipher_args).decode('utf-8')

def create_user_info(request, client, user, scope_set, id_token=False, profile=None):

        user_info['sub'] = make_sub(client, user, profile=profile)

    if profile:

        for attr, userinfo_key in app_settings.PROFILE_OVERRIDE_MAPPING.items():

            if getattr(profile, attr, None) and userinfo_key in user_info:

                user_info[userinfo_key] = getattr(profile, attr)

from authentic2.custom_user.models import Profile

    needs_profile_validation = False

    profile = None

    if request.user.subprofiles.count() and a2_app_settings.A2_USER_PROFILE_MANAGEMENT:

        needs_profile_validation = True

        if (authorized_scopes & scopes) < scopes or needs_profile_validation:

                if request.POST.get('profile-validation', ''):

                    try:

                        profile = Profile.objects.get(

                            user=request.user,

                            id=request.POST['profile-validation'],

                        )

                    except Profile.DoesNotExist:

                        pass

                        'needs_profile_validation': needs_profile_validation,

            profile=profile,

    id_token = utils.create_user_info(

        request,

        client,

        oidc_code.user,

        oidc_code.scope_set(),

        id_token=True,

        profile=oidc_code.profile,

    )

            'sub': utils.make_sub(client, oidc_code.user, profile=oidc_code.profile),

@pytest.fixture

def profile_settings(settings, jwkset):

    settings.A2_IDP_OIDC_JWKSET = jwkset

    settings.A2_IDP_OIDC_PASSWORD_GRANT_RATELIMIT = '100/m'

    settings.A2_USER_PROFILE_MANAGEMENT = True

    return settings

# authentic2 - versatile identity manager

# Copyright (C) 2010-2022 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 pytest

from django.contrib.auth import get_user_model

from django.urls import reverse

from django.utils.encoding import force_text

from django.utils.timezone import now

from jwcrypto.jwk import JWK

from jwcrypto.jwt import JWT

from authentic2.custom_user.models import Profile, ProfileType

from authentic2.utils.misc import make_url

from authentic2_idp_oidc.models import OIDCCode

from authentic2_idp_oidc.utils import make_sub

from .. import utils

from .conftest import client_authentication_headers

User = get_user_model()

pytestmark = pytest.mark.django_db

@pytest.fixture

def profile_user():

    user = User.objects.create(

        first_name='Foo',

        last_name='Bar',

        username='foobar',

        email='foobar@example.org',

    )

    profile_type_manager = ProfileType.objects.create(

        name='One Manager Type',

        slug='one-manager-type',

    )

    profile_type_delegate = ProfileType.objects.create(

        name='One Delegate Type',

        slug='one-delegate-type',

    )

    Profile.objects.create(

        user=user,

        profile_type=profile_type_manager,

        identifier='Entity 789',

        email='manager@example789.org',

    )

    Profile.objects.create(

        user=user,

        profile_type=profile_type_delegate,

        identifier='Entity 1011',

        email='delegate@example1011.org',

    )

    user.clear_password = 'foobar'

    user.set_password('foobar')

    user.save()

    return user

def test_admin_base_models(app, superuser, simple_user, profile_settings):

    url = reverse('admin:custom_user_profiletype_add')

    assert ProfileType.objects.count() == 0

    response = utils.login(app, superuser, path=url)

    response.form.set('name', 'Manager')

    response.form.set('slug', 'manager')

    response = response.form.submit(name='_save').follow()

    assert ProfileType.objects.count() == 1

    response = app.get(url)

    response.form.set('name', 'Delegate')

    response.form.set('slug', 'delegate')

    response = response.form.submit(name='_save').follow()

    assert ProfileType.objects.count() == 2

    url = reverse('admin:custom_user_profile_add')

    assert Profile.objects.count() == 0

    response = app.get(url)

    response.form.set('user', simple_user.id)

    response.form.set('profile_type', ProfileType.objects.first().pk)

    response.form.set('email', 'john.doe@example.org')

    response.form.set('identifier', 'Entity 0123')

    response = response.form.submit(name='_save').follow()

    assert Profile.objects.count() == 1

    response = app.get(url)

    response.form.set('user', simple_user.id)

    response.form.set('profile_type', ProfileType.objects.last().pk)

    response.form.set('email', 'john.doe@anotherexample.org')

    response.form.set('identifier', 'Entity 5678')

    response = response.form.submit(name='_save').follow()

    assert Profile.objects.count() == 2

def test_login_profiles_absent(app, oidc_client, simple_user, profile_settings):

    redirect_uri = oidc_client.redirect_uris.split()[0]

    params = {

        'client_id': oidc_client.client_id,

        'scope': 'openid profile email',

        'redirect_uri': redirect_uri,

        'state': 'xxx',

        'nonce': 'yyy',

        'login_hint': 'backoffice john@example.com',

        'response_type': 'code',

    }

    authorize_url = make_url('oidc-authorize', params=params)

    utils.login(app, simple_user)

    response = app.get(authorize_url)

    assert 'a2-oidc-authorization-form' in response.text

    # not interface changes for users without a profile

    assert not 'profile-validation-' in response.text

def test_login_profile_selection(app, oidc_client, profile_user, profile_settings):

    oidc_client.idtoken_algo = oidc_client.ALGO_HMAC

    oidc_client.save()

    redirect_uri = oidc_client.redirect_uris.split()[0]

    params = {

        'client_id': oidc_client.client_id,

        'scope': 'openid profile email',

        'redirect_uri': redirect_uri,

        'state': 'xxx',

        'nonce': 'yyy',

        'login_hint': 'backoffice john@example.com',

        'response_type': 'code',

    }

    assert profile_user.subprofiles.count() == 2

    authorize_url = make_url('oidc-authorize', params=params)

    utils.login(app, profile_user)

    response = app.get(authorize_url)

    assert 'a2-oidc-authorization-form' in response.text

    assert 'profile-validation-' in response.text

    response.form.set('profile-validation', profile_user.subprofiles.first().id)

    response = response.form.submit('accept')

    assert OIDCCode.objects.count() == 1

    code = OIDCCode.objects.get()

    assert code.client == oidc_client

    assert code.user == profile_user

    assert code.profile == profile_user.subprofiles.first()

    assert code.scope_set() == set('openid profile email'.split())

    assert code.state == 'xxx'

    assert code.nonce == 'yyy'

    assert code.redirect_uri == redirect_uri

    assert code.session_key == app.session.session_key

    assert code.auth_time <= now()

    assert code.expired >= now()

    assert response['Location'].startswith(redirect_uri)

    location = urllib.parse.urlparse(response['Location'])

    query = urllib.parse.parse_qs(location.query)

    assert set(query.keys()) == {'code', 'state'}

    assert query['code'] == [code.uuid]

    code = query['code'][0]

    assert query['state'] == ['xxx']

    token_url = make_url('oidc-token')

    response = app.post(

        token_url,

        params={

            'grant_type': 'authorization_code',

            'code': code,

            'redirect_uri': oidc_client.redirect_uris.split()[0],

        },

        headers=client_authentication_headers(oidc_client),

    )

    assert 'error' not in response.json

    assert 'access_token' in response.json

    assert 'expires_in' in response.json

    assert 'id_token' in response.json

    assert response.json['token_type'] == 'Bearer'

    access_token = response.json['access_token']

    assert access_token

    id_token = response.json['id_token']

    k = base64.b64encode(oidc_client.client_secret.encode('utf-8'))

    key = JWK(kty='oct', k=force_text(k))

    algs = ['HS256']

    jwt = JWT(jwt=id_token, key=key, algs=algs)

    claims = json.loads(jwt.claims)

    # check subject identifier substitution:

    assert claims['sub'] != make_sub(oidc_client, profile_user)

    assert claims['sub'] == make_sub(oidc_client, profile_user, profile=profile_user.subprofiles.first())

    # check email substitution

    assert claims['email'] != profile_user.email

    assert claims['email'] == profile_user.subprofiles.first().email