Produits Entr'ouvert » Authentic 2

    list_display = ['profile_type', 'user', 'identifier', 'email']

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

    list_display = [

        'name',

        'slug',

        'client_id',

        'ou',

        'identifier_policy',

        'created',

        'modified',

        'activate_user_profiles',

        'perform_sub_profile_substitution',

    ]

    @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-03-04 06:38

import django.db.models.deletion

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [

        ('custom_user', '0031_profile_email'),

        ('authentic2_idp_oidc', '0014_auto_20201126_1812'),

    ]

    operations = [

        migrations.AddField(

            model_name='oidcauthorization',

            name='profile',

            field=models.ForeignKey(

                null=True,

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

                to='custom_user.Profile',

                verbose_name='profile',

            ),

        ),

        migrations.AddField(

            model_name='oidcclient',

            name='activate_user_profiles',

            field=models.BooleanField(

                blank=True, default=False, verbose_name="activate users' juridical entity profiles management"

            ),

        ),

        migrations.AddField(

            model_name='oidcclient',

            name='perform_sub_profile_substitution',

            field=models.BooleanField(

                blank=True,

                default=False,

                verbose_name="make pseudonym subs depend on the user's chosen profile",

            ),

        ),

        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

    activate_user_profiles = models.BooleanField(

        verbose_name=_("activate users' juridical entity profiles management"), blank=True, default=False

    )

    perform_sub_profile_substitution = models.BooleanField(

        verbose_name=_("make pseudonym subs depend on the user's chosen profile"),

        blank=True,

        default=False,

    )

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

    profile = models.ForeignKey(

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

    )

#profile-validation {

	background: #f5f5f7;

	padding-top: .3em;

	margin-bottom: .3em;

	border-radius: 4px;

	color: black;

}

#profile-validation > .profile {

	padding: .5em;

	margin-top: .2em;

	margin-bottom: .2em;

	background: #d9d9d9;

}

{% 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_scope_validation %}

    {% if needs_profile_validation %}

     <div id="profile-validation">

     <p>{% trans "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 }}">

          {% with identifier=profile.identifier type_name=profile.profile_type.name %}

            {% trans "Unknown entity" as fallback_identifier %}

            {% blocktrans %}Profile of type {{ type_name }}: {% endblocktrans %}

            {% firstof identifier fallback_identifier %}<br />

          {% endwith %}

          {% if profile.email %}

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

          {% endif %}</label>

        </div>

      {% endfor %}

     </div>

    {% endif %}

def flatten_dict(data):

    for key, value in list(data.items()):

        if isinstance(value, dict):

            flatten_dict(value)

            for key2, value2 in value.items():

                data['%s_%s' % (key, key2)] = value2

            del data[key]

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 and client.perform_sub_profile_substitution:

        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 and client.perform_sub_profile_substitution:

        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)

        user_info['profile_identifier'] = profile.identifier

        user_info['profile_type'] = profile.profile_type.slug

        if isinstance(profile.data, dict):

            flat_data = profile.data.copy()

            flatten_dict(flat_data)

            user_info.update(flat_data)

from authentic2.custom_user.models import Profile

    needs_profile_validation = False

    needs_scope_validation = False

    profile = None

        authorized_profile = None

            # load first authorized profile

            if not authorized_profile and authorization.profile:

                authorized_profile = authorization.profile

        if (

            request.user.subprofiles.count()

            and not authorized_profile

            and client.activate_user_profiles

            and a2_app_settings.A2_USER_PROFILE_MANAGEMENT

        ):

            needs_profile_validation = True

        else:

            profile = authorized_profile

            needs_scope_validation = True

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

                            profile=profile,

                        'needs_profile_validation': needs_profile_validation,

                        'needs_scope_validation': needs_scope_validation,

            profile=profile,

        if 'profile-validation' in request.POST:

            try:

                profile = Profile.objects.get(

                    id=request.POST.get('profile-validation', None),

                    user=request.user,

                )

            except Profile.DoesNotExist:

                pass

        id_token = utils.create_user_info(

            request, client, request.user, scopes, id_token=True, 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 OIDCAccessToken, 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',

    )

    data = {

        'entity_name': 'Foobar',

        'entity_data': {'au': 'ie', 'ts': 'rn'},

    }

    Profile.objects.create(

        user=user,

        profile_type=profile_type_delegate,

        identifier='Entity 1011',

        email='delegate@example1011.org',

        data=data,

    )

    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]

    oidc_client.activate_user_profiles = True

    oidc_client.save()

    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

@pytest.mark.parametrize('sub_profile_substitution', [True, False])

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

    oidc_client.idtoken_algo = oidc_client.ALGO_HMAC

    oidc_client.activate_user_profiles = True

    oidc_client.perform_sub_profile_substitution = sub_profile_substitution

    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:

    if sub_profile_substitution:

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

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

    else:

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

    # check email substitution

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

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

    # check additional profile claims

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

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

    # check profile data dict flatten into oidc claims

    assert claims['entity_name'] == 'Foobar'

    assert claims['entity_data_au'] == 'ie'

    assert claims['entity_data_ts'] == 'rn'

@pytest.mark.parametrize('sub_profile_substitution', [True, False])

def test_login_implicit(app, oidc_client, profile_user, profile_settings, sub_profile_substitution):

    oidc_client.idtoken_algo = oidc_client.ALGO_HMAC

    oidc_client.authorization_flow = oidc_client.FLOW_IMPLICIT

    oidc_client.activate_user_profiles = True

    oidc_client.perform_sub_profile_substitution = sub_profile_substitution

    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': 'token id_token',

    }

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

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

    assert location.fragment

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

    assert OIDCAccessToken.objects.count() == 1

    access_token = OIDCAccessToken.objects.get()

    assert set(query.keys()) == {'access_token', 'token_type', 'expires_in', 'id_token', 'state'}

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

    assert query['token_type'] == ['Bearer']

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

    access_token = query['access_token'][0]

    id_token = query['id_token'][0]

    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:

    if sub_profile_substitution:

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

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

    else:

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

    # check email substitution

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

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

    # check additional profile claims

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

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

    # check profile data dict flatten into oidc claims

    assert claims['entity_name'] == 'Foobar'

    assert claims['entity_data_au'] == 'ie'

    assert claims['entity_data_ts'] == 'rn'