0003-auth_oidc-migrate-authenticator-to-database-53902.patch

Valentin Deniaud, 19 mai 2022 13:46

Voir les différences: en ligne côte à côte

Subject: [PATCH 3/3] auth_oidc: migrate authenticator to database (#53902)

 src/authentic2/settings.py                    |   1 -
 src/authentic2/views.py                       |   2 +-
 src/authentic2_auth_oidc/authenticators.py    |  64 -----------
 src/authentic2_auth_oidc/forms.py             |  32 ++++++
 .../migrations/0001_initial.py                |   3 +
 ...0009_oidcprovider_baseauthenticator_ptr.py |  19 ++++
 .../migrations/0010_auto_20220413_1622.py     |  43 ++++++++
 .../migrations/0011_auto_20220413_1632.py     |  65 +++++++++++
 src/authentic2_auth_oidc/models.py            |  52 ++++++---
 src/authentic2_auth_oidc/utils.py             |  17 ---
 tests/conftest.py                             |   4 +-
 tests/test_auth_oidc.py                       | 101 +++++++++++-------
 tests/test_commands.py                        |   2 +
 tests/test_manager_authenticators.py          |  70 ++++++++++++
 14 files changed, 339 insertions(+), 136 deletions(-)
 delete mode 100644 src/authentic2_auth_oidc/authenticators.py
 create mode 100644 src/authentic2_auth_oidc/forms.py
 create mode 100644 src/authentic2_auth_oidc/migrations/0009_oidcprovider_baseauthenticator_ptr.py
 create mode 100644 src/authentic2_auth_oidc/migrations/0010_auto_20220413_1622.py
 create mode 100644 src/authentic2_auth_oidc/migrations/0011_auto_20220413_1632.py

src/authentic2/settings.py
189	189	AUTH_USER_MODEL = 'custom_user.User'
190	190	AUTH_FRONTENDS = (
191	191	'authentic2_auth_saml.authenticators.SAMLAuthenticator',
192		'authentic2_auth_oidc.authenticators.OIDCAuthenticator',
193	192	'authentic2_auth_fc.authenticators.FcAuthenticator',
194	193	)
195	194

             if hasattr(authenticator, 'autorun'):
                 if 'message' in token:
                     messages.info(request, token['message'])
                 return authenticator.autorun(request, block['id'])
                 return authenticator.autorun(request, block.get('id'))
         # Old frontends API
         for block in blocks:

     # authentic2 - versatile identity manager
     # Copyright (C) 2010-2019 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/>.
     from django.shortcuts import render
     from django.utils.translation import gettext_noop
     from authentic2.authenticators import BaseAuthenticator
     from authentic2.utils.misc import make_url, redirect_to_login
     from . import app_settings, utils
     from .models import OIDCProvider
     class OIDCAuthenticator(BaseAuthenticator):
         id = 'oidc'
         how = ['oidc']
         priority = 2
         def enabled(self):
             return app_settings.ENABLE and utils.has_providers()
         def name(self):
             return gettext_noop('OpenIDConnect')
         def instances(self, request, *args, **kwargs):
             for p in utils.get_providers(shown=True):
                 yield (p.slug, p)
         def autorun(self, request, block_id):
             auth_id, instance_slug = block_id.split('_')
             assert auth_id == self.id
             try:
                 provider = OIDCProvider.objects.get(slug=instance_slug)
             except OIDCProvider.DoesNotExist():
                 return redirect_to_login(request)
             return redirect_to_login(request, login_url='oidc-login', kwargs={'pk': provider.pk})
         def login(self, request, *args, **kwargs):
             context = kwargs.get('context', {})
             if kwargs.get('instance'):
                 instance = kwargs['instance']
                 context['provider'] = instance
                 context['login_url'] = make_url(
                     'oidc-login', kwargs={'pk': instance.id}, request=request, keep_params=True
+                )
                 template_names = [
                     'authentic2_auth_oidc/login_%s.html' % instance.slug,
                     'authentic2_auth_oidc/login.html',
+                ]
                 return render(request, template_names, context)

     # 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/>.
     from django import forms
     from authentic2.apps.authenticators.forms import AuthenticatorFormMixin
     from .models import OIDCProvider
     class OIDCProviderEditForm(AuthenticatorFormMixin, forms.ModelForm):
         class Meta:
             model = OIDCProvider
             fields = '__all__'
         def __init__(self, *args, **kwargs):
             super().__init__(*args, **kwargs)
             self.fields['ou'].required = True
             self.fields['ou'].empty_label = None

                         ),
                     ),
                 ],
                 options={
                     'verbose_name': 'OpenIDConnect',
                 },
             ),
             migrations.AddField(
                 model_name='oidcclaimmapping',

     # Generated by Django 2.2.28 on 2022-04-13 14:22
     from django.db import migrations, models
     class Migration(migrations.Migration):
         dependencies = [
             ('authentic2_auth_oidc', '0008_auto_20201102_1142'),
+        ]
         operations = [
             migrations.AddField(
                 model_name='oidcprovider',
                 name='baseauthenticator_ptr',
                 field=models.IntegerField(default=0),
                 preserve_default=False,
             ),
+        ]

     # Generated by Django 2.2.28 on 2022-04-13 14:22
     from django.db import migrations
     from django.utils.text import slugify
     from authentic2 import app_settings as global_settings
     from authentic2_auth_oidc import app_settings
     def add_base_authenticators(apps, schema_editor):
         kwargs_settings = getattr(global_settings, 'AUTH_FRONTENDS_KWARGS', {})
         oidc_provider_settings = kwargs_settings.get('oidc', {})
         show_condition = oidc_provider_settings.get('show_condition')
         BaseAuthenticator = apps.get_model('authenticators', 'BaseAuthenticator')
         OIDCProvider = apps.get_model('authentic2_auth_oidc', 'OIDCProvider')
         for provider in OIDCProvider.objects.all():
             if isinstance(show_condition, dict):
                 show_condition = show_condition.get(provider.slug, '')
             base_authenticator = BaseAuthenticator.objects.create(
                 name=provider.name,
                 slug=provider.slug or slugify(provider.name),
                 ou=provider.ou,
                 enabled=provider.show and app_settings.ENABLE,
                 order=oidc_provider_settings.get('priority', 2),
                 show_condition=show_condition,
+            )
             provider.baseauthenticator_ptr = base_authenticator.pk
             provider.save()
     class Migration(migrations.Migration):
         dependencies = [
             ('authentic2_auth_oidc', '0009_oidcprovider_baseauthenticator_ptr'),
             ('authenticators', '0001_initial'),
+        ]
         operations = [
             migrations.RunPython(add_base_authenticators, reverse_code=migrations.RunPython.noop),
+        ]

     # Generated by Django 2.2.28 on 2022-04-13 14:32
     import django.db.models.deletion
     from django.db import migrations, models
     from django.db.migrations.operations.base import Operation
     class AlterModelBase(Operation):
         def __init__(self, model_name, base_name):
             self.model_name = model_name
             self.base_name = base_name
         def state_forwards(self, app_label, state):
             model_state = state.models[app_label, self.model_name]
             model_state.bases = (self.base_name,)
             state.reload_model(app_label, self.model_name, delay=True)
         def database_forwards(self, app_label, schema_editor, from_state, to_state):
             pass
         def database_backwards(self, app_label, schema_editor, from_state, to_state):
             pass
     class Migration(migrations.Migration):
         dependencies = [
             ('authentic2_auth_oidc', '0010_auto_20220413_1622'),
+        ]
         operations = [
             migrations.RemoveField(
                 model_name='oidcprovider',
                 name='id',
             ),
             migrations.RemoveField(
                 model_name='oidcprovider',
                 name='name',
             ),
             migrations.RemoveField(
                 model_name='oidcprovider',
                 name='ou',
             ),
             migrations.RemoveField(
                 model_name='oidcprovider',
                 name='show',
             ),
             migrations.RemoveField(
                 model_name='oidcprovider',
                 name='slug',
             ),
             migrations.AlterField(
                 model_name='oidcprovider',
                 name='baseauthenticator_ptr',
                 field=models.OneToOneField(
                     auto_created=True,
                     on_delete=django.db.models.deletion.CASCADE,
                     parent_link=True,
                     primary_key=True,
                     serialize=False,
                     to='authenticators.BaseAuthenticator',
                 ),
             ),
             AlterModelBase('oidcprovider', 'authenticators.baseauthenticator'),
+        ]

     from django.contrib.postgres.fields import JSONField
     from django.core.exceptions import ValidationError
     from django.db import models
     from django.shortcuts import render
     from django.utils.translation import gettext_lazy as _
     from jwcrypto.jwk import InvalidJWKValue, JWKSet
     from authentic2.a2_rbac.models import OrganizationalUnit
     from authentic2.apps.authenticators.models import BaseAuthenticator
     from authentic2.utils.misc import make_url, redirect_to_login
     from authentic2.utils.template import validate_template
     from . import managers
-...
             raise ValidationError(_('Invalid JWKSet: %s') % e)
     class OIDCProvider(models.Model):
     class OIDCProvider(BaseAuthenticator):
         STRATEGY_CREATE = 'create'
         STRATEGY_FIND_UUID = 'find-uuid'
         STRATEGY_FIND_USERNAME = 'find-username'
-...
             (ALGO_EC, _('EC')),
+        ]
         name = models.CharField(unique=True, max_length=128, verbose_name=_('name'))
         slug = models.SlugField(unique=True, max_length=256, verbose_name=_('slug'), blank=True, null=True)
         issuer = models.CharField(max_length=256, verbose_name=_('issuer'), unique=True, db_index=True)
         client_id = models.CharField(max_length=128, default=uuid.uuid4, verbose_name=_('client id'))
         client_secret = models.CharField(max_length=128, default=uuid.uuid4, verbose_name=_('client secret'))
-...
         # ou where new users should be created
         strategy = models.CharField(max_length=32, choices=STRATEGIES, verbose_name=_('strategy'))
         ou = models.ForeignKey(
             to=OrganizationalUnit, verbose_name=_('organizational unit'), on_delete=models.CASCADE
+        )
         # policy
         max_auth_age = models.PositiveIntegerField(
             verbose_name=_('max authentication age'), blank=True, null=True
+        )
         # hide OP from login page
         show = models.BooleanField(verbose_name=_('show on login page'), blank=True, default=True)
         # metadata
         created = models.DateTimeField(verbose_name=_('created'), auto_now_add=True)
         modified = models.DateTimeField(verbose_name=_('modified'), auto_now=True)
         objects = managers.OIDCProviderManager()
         type = 'oidc'
         how = ['oidc']
         description_fields = ['show_condition', 'issuer', 'scopes', 'strategy', 'created', 'modified']
         class Meta:
             verbose_name = _('OpenIDConnect')
         @property
         def manager_form_class(self):
             from .forms import OIDCProviderEditForm
             return OIDCProviderEditForm
         @property
         def jwkset(self):
             if self.jwkset_json:
                 return JWKSet.from_json(json.dumps(self.jwkset_json))
             return None
         def get_short_description(self):
             if self.issuer and self.scopes:
                 return _('OIDC provider linked to issuer %(issuer)s with scopes %(scopes)s.') % {
                     'issuer': self.issuer,
                     'scopes': self.scopes.replace(' ', ', '),
+                }
         def clean_fields(self, exclude=None):
             super().clean_fields(exclude=exclude)
             exclude = exclude or []
-...
                         % key_sig_mapping[self.idtoken_algo]
+                    )
         def __str__(self):
             return str(self.name)
         def authorization_claims_parameter(self):
             idtoken_claims = {}
             userinfo_claims = {}
-...
         def __repr__(self):
             return '<OIDCProvider %r>' % self.issuer
         def autorun(self, request, *args):
             return redirect_to_login(request, login_url='oidc-login', kwargs={'pk': self.pk})
         def login(self, request, *args, **kwargs):
             context = kwargs.get('context', {})
             context['provider'] = self
             context['login_url'] = make_url(
                 'oidc-login', kwargs={'pk': self.id}, request=request, keep_params=True
+            )
             template_names = [
                 'authentic2_auth_oidc/login_%s.html' % self.slug,
                 'authentic2_auth_oidc/login.html',
+            ]
             return render(request, template_names, context)
     class OIDCClaimMapping(models.Model):
         NOT_VERIFIED = 0

     from authentic2.models import Attribute
     from authentic2.utils.cache import GlobalCache
     from . import models
     TIMEOUT = 1
     @GlobalCache(timeout=5, kwargs=['shown'])
     def get_providers(shown=None):
         qs = models.OIDCProvider.objects.all()
         if shown is not None:
             qs = qs.filter(show=shown)
         return qs
     @GlobalCache(timeout=TIMEOUT)
     def get_attributes():
         return Attribute.objects.all()
-...
         return get_object_or_404(models.OIDCProvider, pk=pk)
     @GlobalCache(timeout=TIMEOUT)
     def has_providers():
         from . import models
         return models.OIDCProvider.objects.filter(show=True).exists()
     @GlobalCache(timeout=TIMEOUT)
     def get_provider_by_issuer(issuer):
         from . import models

     from authentic2.manager.utils import get_ou_count
     from authentic2.models import Attribute, Service
     from authentic2.utils.evaluate import BaseExpressionValidator
     from authentic2_auth_oidc.utils import get_provider_by_issuer, get_providers, has_providers
     from authentic2_auth_oidc.utils import get_provider_by_issuer
     from authentic2_idp_oidc.models import OIDCClient
     from . import utils
-...
         for cached_el in (
             OrganizationalUnit.cached,
             a2_hooks.get_hooks,
             get_providers,
             get_provider_by_issuer,
             get_ou_count,
             has_providers,
         ):
             cached_el.cache.clear()

     from authentic2.utils.misc import last_authentication_event
     from authentic2_auth_oidc.backends import OIDCBackend
     from authentic2_auth_oidc.models import OIDCAccount, OIDCClaimMapping, OIDCProvider
     from authentic2_auth_oidc.utils import (
         IDToken,
         IDTokenError,
         get_providers,
         has_providers,
         parse_id_token,
         register_issuer,
+    )
     from authentic2_auth_oidc.utils import IDToken, IDTokenError, parse_id_token, register_issuer
     from . import utils
-...
             ou=get_default_ou(),
             name=name,
             slug=slug,
             enabled=True,
             issuer=issuer,
             authorization_endpoint='%s/authorize' % issuer,
             token_endpoint='%s/token' % issuer,
-...
             ou=get_default_ou(),
             name='OIDCIDP 2',
             slug='oidcidp-2',
             enabled=True,
             issuer='https://idp2.example.com/',
             authorization_endpoint='https://idp2.example.com/authorize',
             token_endpoint='https://idp2.example.com/token',
-...
     def test_login_with_conditional_authenticators(oidc_provider, oidc_provider_jwkset, app, settings, caplog):
         make_oidc_provider(name='My IDP', slug='myidp', jwkset=oidc_provider_jwkset)
         myidp = make_oidc_provider(name='My IDP', slug='myidp', jwkset=oidc_provider_jwkset)
         response = app.get('/login/')
         assert 'My IDP' in response
         assert 'Server' in response
         settings.AUTH_FRONTENDS_KWARGS = {'oidc': {'show_condition': {'myidp': 'remote_addr==\'0.0.0.0\''}}}
         myidp.show_condition = 'remote_addr==\'0.0.0.0\''
         myidp.save()
         response = app.get('/login/')
         assert 'Server' in response
         assert 'My IDP' not in response
         settings.AUTH_FRONTENDS_KWARGS = {
             'oidc': {
                 'show_condition': {'myid': 'remote_addr==\'0.0.0.0\'', 'server': 'remote_addr==\'127.0.0.1\''}
+            }
+        }
         response = app.get('/login/')
         assert 'Server' in response
         assert 'My IDP' in response
         settings.AUTH_FRONTENDS_KWARGS = {
             'oidc': {
                 'show_condition': {'myidp': 'remote_addr==\'0.0.0.0\'', 'server': 'remote_addr==\'127.0.0.1\''}
+            }
+        }
         oidc_provider.show_condition = 'remote_addr==\'127.0.0.1\''
         oidc_provider.save()
         response = app.get('/login/')
         assert 'Server' in response
         assert 'My IDP' not in response
         settings.AUTH_FRONTENDS_KWARGS = {'oidc': {'show_condition': 'remote_addr==\'127.0.0.1\''}}
         myidp.show_condition = 'remote_addr==\'127.0.0.1\''
         myidp.save()
         response = app.get('/login/')
         assert 'Server' in response
         assert 'My IDP' in response
         settings.AUTH_FRONTENDS_KWARGS = {
             'oidc': {
                 'show_condition': {
                     'myidp': 'remote_addr==\'127.0.0.1\' and \'backoffice\' not in login_hint',
                     'server': '\'backoffice\' in login_hint',
+                }
+            }
+        }
         myidp.show_condition = 'remote_addr==\'127.0.0.1\' and \'backoffice\' not in login_hint'
         myidp.save()
         oidc_provider.show_condition = '\'backoffice\' in login_hint'
         oidc_provider.save()
         response = app.get('/login/')
         assert 'Server' not in response
         assert 'My IDP' in response
-...
         assert 'oidc-a-server' in response.text
         # do not show this provider on login page anymore
         oidc_provider.show = False
         oidc_provider.enabled = False
         oidc_provider.save()
         # we have a 5 seconds cache on list of providers, we have to work around it
         get_providers.cache.clear()
         has_providers.cache.clear()
         response = app.get('/login/')
         assert 'oidc-a-server' not in response.text
-...
             ):
                 response = app.get(login_callback_url(oidc_provider), params={'code': code, 'state': state})
             assert User.objects.count() == 1
     @pytest.mark.parametrize(
         'auth_frontend_kwargs',
+        [
             {'oidc': {'priority': 3, 'show_condition': '"backoffice" not in login_hint'}},
             {'oidc': {'show_condition': {'baz': '"backoffice" not in login_hint', 'bar': 'True'}}},
         ],
+    )
     def test_oidc_provider_authenticator_data_migration(auth_frontend_kwargs, migration, settings):
         settings.AUTH_FRONTENDS_KWARGS = auth_frontend_kwargs
         app = 'authentic2_auth_oidc'
         migrate_from = [(app, '0008_auto_20201102_1142')]
         migrate_to = [(app, '0011_auto_20220413_1632')]
         old_apps = migration.before(migrate_from)
         OIDCProvider = old_apps.get_model(app, 'OIDCProvider')
         OrganizationalUnit = old_apps.get_model('a2_rbac', 'OrganizationalUnit')
         ou1 = OrganizationalUnit.objects.create(name='OU1', slug='ou1')
         issuer = 'https://baz.example.com'
         OIDCProvider.objects.create(
             name='Baz',
             slug='baz',
             ou=ou1,
             show=True,
             issuer=issuer,
             authorization_endpoint='%s/authorize' % issuer,
             token_endpoint='%s/token' % issuer,
             end_session_endpoint='%s/logout' % issuer,
             userinfo_endpoint='%s/user_info' % issuer,
             token_revocation_endpoint='%s/revoke' % issuer,
+        )
         new_apps = migration.apply(migrate_to)
         OIDCProvider = new_apps.get_model(app, 'OIDCProvider')
         BaseAuthenticator = new_apps.get_model('authenticators', 'BaseAuthenticator')
         authenticator = OIDCProvider.objects.get()
         assert authenticator.name == 'Baz'
         assert authenticator.slug == 'baz'
         assert authenticator.ou.pk == ou1.pk
         assert authenticator.enabled is True
         assert authenticator.order == auth_frontend_kwargs['oidc'].get('priority', 2)
         assert authenticator.show_condition == '"backoffice" not in login_hint'
         assert authenticator.authorization_endpoint == '%s/authorize' % issuer
         base_authenticator = BaseAuthenticator.objects.get()
         assert authenticator.uuid == base_authenticator.uuid

             jwkset.add(JWK.generate(kty='EC', size=256, kid='tsrn'))
             return OIDCProvider.objects.create(
                 name=name,
                 slug='test',
                 enabled=True,
                 ou=ou,
                 issuer=issuer,
                 strategy='create',

     # 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 pytest
     from authentic2_auth_oidc.models import OIDCProvider
     from .utils import login, logout
-...
         # cannot add another password authenticator
         resp = app.get('/manage/authenticators/add/')
         assert 'Password' not in resp.text
     @pytest.mark.freeze_time('2022-04-19 14:00')
     def test_authenticators_oidc(app, superuser, ou1, ou2):
         resp = login(app, superuser, path='/manage/authenticators/')
         resp = resp.click('Add new authenticator')
         resp.form['name'] = 'Test'
         resp.form['authenticator'] = 'oidc'
         resp.form['ou'] = ou1.pk
         resp = resp.form.submit().follow()
         assert OIDCProvider.objects.filter(slug='test').count() == 1
         assert 'Created: April 19, 2022, 2 p.m.' in resp.text
         assert 'Modified: April 19, 2022, 2 p.m.' in resp.text
         assert 'Issuer' not in resp.text
         resp = resp.click('Edit')
         assert 'enabled' not in resp.form.fields
         resp.form['issuer'] = 'https://oidc.example.com'
         resp.form['scopes'] = 'profile email'
         resp.form['strategy'] = 'create'
         resp.form['authorization_endpoint'] = 'https://oidc.example.com/authorize'
         resp.form['token_endpoint'] = 'https://oidc.example.com/token'
         resp.form['userinfo_endpoint'] = 'https://oidc.example.com/user_info'
         resp.form['idtoken_algo'] = 2
         resp = resp.form.submit().follow()
         assert 'Issuer: https://oidc.example.com' in resp.text
         assert 'Scopes: profile email' in resp.text
         resp = app.get('/manage/authenticators/')
         assert 'OpenIDConnect - Test' in resp.text
         assert 'class="section disabled"' in resp.text
         assert 'OIDC provider linked to' not in resp.text
         resp = resp.click('Configure', index=1)
         resp = resp.click('Enable').follow()
         assert 'Authenticator has been enabled.' in resp.text
         resp = app.get('/manage/authenticators/')
         assert 'class="section disabled"' not in resp.text
         assert 'OIDC provider linked to https://oidc.example.com with scopes profile, email.' not in resp.text
         # same name
         resp = resp.click('Add new authenticator')
         resp.form['name'] = 'test'
         resp.form['authenticator'] = 'oidc'
         resp.form['ou'] = ou1.pk
         resp = resp.form.submit().follow()
         assert OIDCProvider.objects.filter(slug='test-1').count() == 1
         OIDCProvider.objects.filter(slug='test-1').delete()
         # OU is required
         resp = app.get('/manage/authenticators/add/')
         resp.form['name'] = 'test'
         resp.form['authenticator'] = 'oidc'
         resp.form['ou'] = ''
         resp = resp.form.submit()
         assert 'This field is required' in resp.text
         resp = app.get('/manage/authenticators/')
         resp = resp.click('Configure', index=1)
         resp = resp.click('Delete')
         resp = resp.form.submit().follow()
         assert not OIDCProvider.objects.filter(slug='test').exists()
+    -

Projet

Général

Profil

Produits Entr'ouvert » Authentic 2

0003-auth_oidc-migrate-authenticator-to-database-53902.patch