Produits Entr'ouvert » Authentic 2

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

import requests

from django.core.exceptions import MultipleObjectsReturned

from django.core.management.base import BaseCommand

from authentic2.utils.template import Template

from authentic2_auth_oidc.models import OIDCAccount, OIDCProvider

class Command(BaseCommand):

    def add_arguments(self, parser):

        parser.add_argument('--delta', metavar='DELTA', type=int, default=300)

        parser.add_argument('--provider', type=str, default=None)

    def handle(self, *args, **options):

        verbose = int(options['verbosity']) > 0

        delta = options['delta']

        provider = options['provider']

        if not provider:

            self.stdout.write(self.style.ERROR('no declared provider, exiting...'))

            return

        try:

            provider = OIDCProvider.objects.get(slug=provider)

        except OIDCProvider.DoesNotExist:

            self.stdout.write(self.style.ERROR(f'provider {provider} not found, exiting...'))

            return

        # check all existing users

        def chunks(l, n):

            for i in range(0, len(l), n):

                yield l[i : i + n]

        url = provider.issuer + '/api/users/synchronization/'

        unknown_uuids = []

        auth = (provider.client_id, provider.client_secret)

        for accounts in chunks(OIDCAccount.objects.filter(provider=provider), 100):

            subs = [x.sub for x in accounts]

            resp = requests.post(url, json={'known_uuids': subs}, auth=auth)

            resp.raise_for_status()

            unknown_uuids.extend(resp.json().get('unknown_uuids'))

        deletion_ratio = len(unknown_uuids) / OIDCAccount.objects.filter(provider=provider).count()

        if deletion_ratio > 0.05:  # higher than 5%, something definitely went wrong

            self.stdout.write(

                self.style.ERROR(

                    f'deletion ratio is abnormally high ({deletion_ratio}), aborting unkwown users deletion'

                )

            )

        else:

            OIDCAccount.objects.filter(sub__in=unknown_uuids).delete()

        # update recently modified users

        url = provider.issuer + '/api/users/?modified__gt=%s&claim_resolution' % (

            datetime.datetime.now() - datetime.timedelta(seconds=delta)

        ).strftime('%Y-%m-%dT%H:%M:%S')

        while url:

            resp = requests.get(url, auth=auth)

            resp.raise_for_status()

            url = resp.json().get('next')

            if verbose:

                self.stdout.write('got %s users' % len(resp.json()['results']))

            for user_dict in resp.json()['results']:

                try:

                    account = OIDCAccount.objects.get(user__email=user_dict['email'])

                except OIDCAccount.DoesNotExist:

                    continue

                except MultipleObjectsReturned:

                    continue

                had_changes = False

                for claim in provider.claim_mappings.all():

                    if '{{' in claim.claim or '{%' in claim.claim:

                        template = Template(claim.claim)

                        attribute_value = template.render(context=user_dict)

                    else:

                        attribute_value = user_dict.get(claim.claim)

                    try:

                        old_attribute_value = getattr(account.user, claim.attribute)

                    except AttributeError:

                        try:

                            old_attribute_value = getattr(account.user.attributes, claim.attribute)

                        except AttributeError:

                            old_attribute_value = None

                    if old_attribute_value == attribute_value:

                        continue

                    had_changes = True

                    setattr(account.user, claim.attribute, attribute_value)

                    try:

                        setattr(account.user.attributes, claim.attribute, attribute_value)

                    except AttributeError:

                        pass

                if had_changes:

                    if verbose:

                        self.stdout.write('had changes, saving %r' % account.user)

                    account.user.save()

import random

import uuid

import httmock

from django.core.management import call_command

from authentic2.utils import crypto

from authentic2_auth_oidc.models import OIDCAccount, OIDCClaimMapping, OIDCProvider

@pytest.mark.parametrize('deletion_number,deletion_valid', [(2, True), (5, True), (10, False)])

def test_oidc_sync_provider(db, app, admin, settings, capsys, deletion_number, deletion_valid):

    oidc_provider = OIDCProvider.objects.create(

        issuer='https://some.provider',

        name='Some Provider',

        slug='some-provider',

        ou=get_default_ou(),

    )

    OIDCClaimMapping.objects.create(

        provider=oidc_provider,

        attribute='username',

        idtoken_claim=False,

        claim='username',

    )

    OIDCClaimMapping.objects.create(

        provider=oidc_provider,

        attribute='email',

        idtoken_claim=False,

        claim='email',

    )

    # last one, with an idtoken claim

    OIDCClaimMapping.objects.create(

        provider=oidc_provider,

        attribute='last_name',

        idtoken_claim=True,

        claim='family_name',

    )

    # typo in template string

    OIDCClaimMapping.objects.create(

        provider=oidc_provider,

        attribute='first_name',

        idtoken_claim=True,

        claim='given_name',

    )

    User = get_user_model()

    for i in range(100):

        user = User.objects.create(

            first_name='John%s' % i,

            last_name='Doe%s' % i,

            username='john.doe.%s' % i,

            email='john.doe.%s@ad.dre.ss' % i,

            ou=get_default_ou(),

        )

        identifier = uuid.UUID(user.uuid).bytes

        sector_identifier = 'some.provider'

        cipher_args = [

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

            identifier,

            sector_identifier,

        ]

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

        OIDCAccount.objects.create(user=user, provider=oidc_provider, sub=sub)

    def synchronization_post_deletion_response(url, request):

        headers = {'content-type': 'application/json'}

        content = {

            'unknown_uuids': [

                account.sub for account in random.sample(list(OIDCAccount.objects.all()), deletion_number)

            ]

        }

        return httmock.response(status_code=200, headers=headers, content=content, request=request)

    def synchronization_get_modified_response(url, request):

        headers = {'content-type': 'application/json'}

        # randomized batch of modified users

        modified_users = random.sample(list(User.objects.all()), 20)

        results = []

        for count, user in enumerate(modified_users):

            user_json = user.to_json()

            user_json['username'] = f'modified_{count}'

            user_json['first_name'] = 'Mod'

            user_json['last_name'] = 'Ified'

            # mocking claim resolution by oidc provider

            user_json['given_name'] = 'Mod'

            user_json['family_name'] = 'Ified'

            results.append(user_json)

        content = {'results': results}

        return httmock.response(status_code=200, headers=headers, content=content, request=request)

    with httmock.HTTMock(

        httmock.urlmatch(

            netloc=r'some\.provider',

            path=r'^/api/users/synchronization/$',

            method='POST',

        )(synchronization_post_deletion_response)

    ):

        with httmock.HTTMock(

            httmock.urlmatch(

                netloc=r'some\.provider',

                path=r'^/api/users/*',

                method='GET',

            )(synchronization_get_modified_response)

        ):

            call_command('oidc-sync-provider', '--delta', '300', '-v1')

            out, err = capsys.readouterr()

            assert 'no declared provider' in out

            call_command('oidc-sync-provider', '--delta', '300', '--provider', 'unknown-provider', '-v1')

            out, err = capsys.readouterr()

            assert 'provider unknown-provider not found' in out

            call_command('oidc-sync-provider', '--delta', '300', '--provider', 'some-provider', '-v1')

            out, err = capsys.readouterr()

            assert not err

            if deletion_valid:

                # existing users check

                assert OIDCAccount.objects.count() == 100 - deletion_number

            else:

                assert 'deletion ratio is abnormally high' in out

                assert OIDCAccount.objects.count() == 100

            # users update

            assert 'got 20 users' in out

            assert User.objects.filter(username__startswith='modified').count() in range(

                20 - deletion_number, 21

            )

            assert User.objects.filter(first_name='Mod', last_name='Ified').count() in range(

                20 - deletion_number, 21

            )