Produits Entr'ouvert » Authentic 2

            else:

                username = form.cleaned_data.get('username', '').strip()

                if username:

                    request.journal.record('user.login.failure', username=username)

# authentic2 - versatile identity manager

# Copyright (C) 2010-2020 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 authentic2.utils.service import get_service_from_request

from authentic2.apps.journal.journal import Journal

class Journal(Journal):

    def __init__(self, **kwargs):

        self._service = kwargs.pop('service', None)

        super().__init__(**kwargs)

    @property

    def service(self):

        return self._service or (get_service_from_request(self.request) if self.request else None)

    def massage_kwargs(self, record_parameters, kwargs):

        if 'service' not in kwargs and 'service' in record_parameters:

            kwargs['service'] = self.service

        return super().massage_kwargs(record_parameters, kwargs)

journal = Journal()

# authentic2 - versatile identity manager

# Copyright (C) 2010-2020 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.utils.translation import ugettext_lazy as _

from authentic2.custom_user.models import get_attributes_map

from authentic2.apps.journal.models import EventTypeDefinition

from authentic2.apps.journal.utils import form_to_old_new

from authentic2.custom_user.models import User

from . import models

class EventTypeWithService(EventTypeDefinition):

    @classmethod

    def record(cls, user=None, service=None, session=None, references=None, data=None):

        if service:

            if not data:

                data = {}

            data['service_name'] = str(service)

            if not references:

                references = []

            references = [service] + references

        super().record(user=user, session=session, references=references, data=data)

    @classmethod

    def get_service_name(self, event):

        service, = event.get_typed_references(models.Service)

        if service is not None:

            return str(service)

        if 'service_name' in event.data:

            return event.data['service_name']

        return ''

def login_method_label(how):

    if how.startswith('password'):

        return _('password')

    elif how == 'fc':

        return _('FranceConnect')

    elif how == 'saml':

        return _('SAML')

    elif how == 'oidc':

        return _('OpenIDConnect')

    elif how:

        return how

    else:

        return _('none')

def get_attributes_label(attributes_new_values):

    # FIXME: attributes cache should be factorized at the level of the AttributeManager as done on ContentType,

    # main difference is that attributes are editable objects

    attributes_map = get_attributes_map()

    for name in attributes_new_values:

        if name in ('email', 'first_name', 'last_name'):

            yield str(User._meta.get_field(name).verbose_name)

        else:

            if name in attributes_map:

                yield attributes_map[name].label

            else:

                yield name

class UserLogin(EventTypeWithService):

    name = 'user.login'

    label = _('login')

    @classmethod

    def record(cls, user, session, service, how):

        super().record(user=user, session=session, service=service, data={'how': how})

    @classmethod

    def get_message(cls, event, context):

        how = event.get_data('how')

        return _('login using {method}').format(method=login_method_label(how))

class UserLoginFailure(EventTypeWithService):

    name = 'user.login.failure'

    label = _('login failure')

    @classmethod

    def record(cls, service, username):

        super().record(service=service, data={'username': username})

    @classmethod

    def get_message(cls, event, context):

        username = event.get_data('username')

        return _('login failure with username "{username}"').format(username=username)

class UserRegistrationRequest(EventTypeDefinition):

    name = 'user.registration.request'

    label = _('registration request')

    @classmethod

    def record(cls, email):

        super().record(data={'email': email.lower()})

    @classmethod

    def get_message(cls, event, context):

        email = event.get_data('email')

        return _('registration request with email "%s"') % email

class UserRegistration(EventTypeWithService):

    name = 'user.registration'

    label = _('registration')

    @classmethod

    def record(cls, user, session, service, how):

        super().record(user=user, session=session, service=service, data={'how': how})

    @classmethod

    def get_message(cls, event, context):

        how = event.get_data('how')

        return _('registration using {method}').format(method=login_method_label(how))

class UserLogout(EventTypeWithService):

    name = 'user.logout'

    label = _('logout')

    @classmethod

    def record(cls, user, session, service):

        super().record(user=user, session=session, service=service)

    @classmethod

    def get_message(cls, event, context):

        return _('logout')

class UserRequestPasswordReset(EventTypeDefinition):

    name = 'user.password.reset.request'

    label = _('password reset request')

    @classmethod

    def record(cls, user, email):

        super().record(user=user, data={'email': email.lower()})

    @classmethod

    def get_message(cls, event, context):

        email = event.get_data('email')

        if email:

            return _('password reset request with email "%s"') % email

        return super().get_message(event, context)

class UserResetPassword(EventTypeDefinition):

    name = 'user.password.reset'

    label = _('password reset')

    @classmethod

    def record(cls, user, session):

        super().record(user=user, session=session)

class UserResetPasswordFailure(EventTypeDefinition):

    name = 'user.password.reset.failure'

    label = _('password reset failure')

    @classmethod

    def record(cls, email):

        super().record(data={'email': email})

    @classmethod

    def get_message(cls, event, context):

        email = event.get_data('email')

        if email:

            return _('password reset failure with email "%s"') % email

        return super().get_message(event, context)

class UserChangePassword(EventTypeWithService):

    name = 'user.password.change'

    label = _('password change')

    @classmethod

    def record(cls, user, session, service):

        super().record(user=user, session=session, service=service)

class UserEdit(EventTypeWithService):

    name = 'user.profile.edit'

    label = _('profile edit')

    @classmethod

    def record(cls, user, session, service, form):

        data = form_to_old_new(form)

        super().record(user=user, session=session, service=service, data=data)

    @classmethod

    def get_message(cls, event, context):

        new = event.get_data('new')

        if new:

            edited_attributes = ', '.join(get_attributes_label(new))

            return _('profile edit (%s)') % edited_attributes

        return super().get_message(event, context)

class UserDeletion(EventTypeWithService):

    name = 'user.deletion'

    label = _('deletion')

    @classmethod

    def record(cls, user, session, service):

        super().record(user=user, session=session, service=service)

class UserServiceSSO(EventTypeWithService):

    name = 'user.service.sso'

    label = _('service single sign on')

    @classmethod

    def record(cls, user, session, service, how):

        super().record(user=user, session=session, service=service, data={'how': how})

    @classmethod

    def get_message(cls, event, context):

        service_name = cls.get_service_name(event)

        return _('service single sign on with "{service}"').format(

            service=service_name)

class UserServiceSSOAuthorization(EventTypeWithService):

    name = 'user.service.sso.authorization'

    label = _('consentment to single sign on')

    @classmethod

    def record(cls, user, session, service, **kwargs):

        super().record(user=user, session=session, service=service, data=kwargs)

    @classmethod

    def get_message(cls, event, context):

        service_name = cls.get_service_name(event)

        return _('authorization of single sign on with "{service}"').format(

            service=service_name)

class UserServiceSSOUnauthorization(EventTypeWithService):

    name = 'user.service.sso.unauthorization'

    label = _('remove consentment to single sign on')

    @classmethod

    def record(cls, user, session, service):

        super().record(user=user, session=session, service=service)

    @classmethod

    def get_message(cls, event, context):

        service_name = cls.get_service_name(event)

        return _('unauthorization of single sign on with "{service}"').format(

            service=service_name)

from django.utils.functional import SimpleLazyObject

from .utils.service import get_service_from_request, get_service_from_session

        request.service = SimpleLazyObject(lambda: get_service_from_session(request))

def journal_middleware(get_response):

    from . import journal

    def middleware(request):

        request.journal = journal.Journal(request=request)

        return get_response(request)

    return middleware

    'authentic2.middleware.journal_middleware',

    request.journal.record('user.login', how=how)

    request.journal.record('user.registration.request', email=email)

    from authentic2.journal import journal

    journal.record('user.password.reset.request', email=user.email, user=user)

    if service_ref and '\x00' not in service_ref:

        return get_service_from_ref(service_ref)

    return None

def get_service_from_session(request):

    session = getattr(request, 'session', None)

    if session and 'service_pk' in session:

        from authentic2.models import Service

        return Service.objects.get(pk=session['service_pk'])

    return None

        self.request.journal.record('user.profile.edit', form=form)

        request.journal.record('user.logout')

            self.request.journal.record('user.password.reset.failure', email=email)

            self.request.journal.record('user.password.reset.failure', email=email)

        response = utils.simulate_authentication(self.request, self.user, 'email')

        self.request.journal.record('user.password.reset')

        return response

        request.journal.record(

            'user.registration',

            user=user,

            session=None,

            how=self.authentication_method)

            request.journal.record('user.deletion', user=self.user)

        response = super(PasswordChangeView, self).form_valid(form)

        self.request.journal.record('user.password.change', session=self.request.session)

        return response

                        request.journal.record(

                            'user.service.sso.authorization',

                            service=client,

                            scopes=list(sorted(scopes)))

    request.journal.record(

        'user.service.sso',

        service=client,

        how=last_auth and last_auth.get('how'))

        user.clear_password = password

    return create_user(username='john.doe.ou2', first_name=u'Jôhn', last_name=u'Dôe',

from .utils import Authentic2TestCase, get_response_form, get_link_from_mail, assert_event

        self.user = user

        new = {'custom': 'random data', 'next_url': '', 'national_number': 'xx20153566342yy'}

        assert_event('user.profile.edit', user=self.user, session=self.client.session, old={}, new=new)

        utils.assert_event('user.service.sso.authorization',

                           session=app.session,

                           user=simple_user, service=oidc_client,

                           scopes=['email', 'openid', 'profile'])

    utils.assert_event('user.service.sso', session=app.session,

                       user=simple_user, service=oidc_client,

                       how='password-on-https')

from .utils import login, check_log, assert_event

User = get_user_model()

def test_success(db, app, simple_user):

    login(app, simple_user)

    assert_event('user.login', user=simple_user, session=app.session, how='password-on-https')

    session = app.session

    app.get('/logout/').form.submit()

    assert_event('user.logout', user=simple_user, session=session)

def test_failure(db, app, simple_user):

    login(app, simple_user, password='wrong', fail=True)

    assert_event('user.login.failure', username=simple_user.username)

    login(app, user2)

    with utils.run_on_commit_hooks():

        send_password_reset_mail(

            simple_user,

            legacy_subject_templates=['registration/password_reset_subject.txt'],

            legacy_body_templates=['registration/password_reset_email.html'],

            context={

                'base_url': 'http://testserver',

            })

    utils.assert_event('user.password.reset.request', user=simple_user, email=simple_user.email)

    utils.assert_event('user.password.reset', user=simple_user, session=app.session)

    utils.assert_event('user.password.reset.request', user=simple_user, email=simple_user.email)

from .utils import get_link_from_mail, assert_event

def test_registration_success(app, db, settings, mailoutbox, external_redirect):

    assert_event('user.registration.request', email='testbot@entrouvert.com')

    assert_event('user.registration', user=new_user, how='email')

from authentic2.journal import Journal

    request.journal = Journal(request=request)

from .utils import login, logout, get_link_from_mail, assert_event

    old_session_key = app.session.session_key

    new_session_key = app.session.session_key

    assert old_session_key != new_session_key, 'session\'s key has not been cycled'

    assert_event('user.password.change', user=simple_user, session=app.session)

    assert 'You are about to delete the account of <strong>%s</strong>.' % escape(

        simple_user.get_full_name()) in page.text

    assert 'You are about to delete the account of <strong>%s</strong>.' % escape(

        simple_user.get_full_name()) in page.text

    response = (

        app.get(reverse('validate_deletion',

                kwargs={'deletion_token': 'thisismostlikelynotavalidtoken'}))

        .follow()

        .follow()

    )

    if view_name == 'password_reset':

        assert_event('user.password.reset.failure', email=simple_user.email)

from authentic2 import utils, models

from authentic2.apps.journal.models import Event

def login(app, user, path=None, password=None, remember_me=None, args=None, kwargs=None, fail=False):

    username = user.username if hasattr(user, 'username') else user

    form.set('username', username)

    form.set('password', password or (user.clear_password if hasattr(user, 'clear_password') else username))

    response = form.submit(name='login-password-submit')

    if fail:

        assert response.status_code == 200

        assert '_auth_user_id' not in app.session

        response = response.follow()

        if path:

            assert response.request.path == path

        else:

            assert response.request.path == reverse('auth_homepage')

        assert '_auth_user_id' in app.session

        assert not hasattr(user, 'id') or (app.session['_auth_user_id'] == str(user.id))

    return u''.join(node.itertext()) if node is not None else ''

def assert_event(event_type_name, user=None, session=None, service=None, **data):

    qs = Event.objects.filter(type__name=event_type_name)

    if user:

        qs = qs.filter(user=user)

    else:

        qs = qs.filter(user__isnull=True)

    if session:

        qs = qs.filter(session=session.session_key)

    else:

        qs = qs.filter(session__isnull=True)

    if service:

        qs = qs.which_references(service)

    else:

        qs = qs.exclude(qs._which_references_query(models.Service))

    assert qs.count() == 1

    if data:

        event = qs.get()

        assert event.data, 'no event.data, should be %s' % data

        for key, value in data.items():

            assert event.data.get(key) == value, (

                'event.data[%s] != data[%s] (%s != %s)' % (key, key, event.data.get(key), value)

            )