Produits Entr'ouvert » Authentic 2

recursive-include src/authentic2_journal/templates/authentic2_journal *.html

recursive-include src/authentic2_journal/locale *.po *.mo

              'authentic2-journal = authentic2_journal:Plugin',

    'authentic2_journal',

# authentic - Versatile identity management server

# Copyright (C) 2018 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/>.

class Plugin(object):

    def get_after_urls(self):

        from django.conf.urls import include, url

        from . import urls

        return [url(r'^manage/', include(urls))]

    def get_apps(self):

        return [__name__]

default_app_config = 'authentic2_journal.apps.Authentic2JournalConfig'

# authentic - Versatile identity management server

# Copyright (C) 2018 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/>.

class AppSettings(object):

    __DEFAULTS = {

            'ENABLE': False,

    }

    def __init__(self, prefix):

        self.prefix = prefix

    def _setting(self, name, dflt):

        from django.conf import settings

        return getattr(settings, self.prefix + name, dflt)

    def __getattr__(self, name):

        if name not in self.__DEFAULTS:

            raise AttributeError(name)

        return self._setting(name, self.__DEFAULTS[name])

import sys

app_settings = AppSettings('A2_JOURNAL_')

app_settings.__name__ = __name__

sys.modules[__name__] = app_settings

# authentic - Versatile identity management server

# Copyright (C) 2018 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 __future__ import unicode_literals

from django.db import DEFAULT_DB_ALIAS, router

from django.apps import AppConfig

class Authentic2JournalConfig(AppConfig):

    name = 'authentic2_journal'

    verbose_name = 'Authentic 2 Journal Application'

    def ready(self):

        from django.db.models.signals import post_migrate

        post_migrate.connect(

            self.create_event_kinds,

            sender=self)

    def create_event_kinds(self, app_config, verbosity=2, interactive=True,

            using=DEFAULT_DB_ALIAS, **kwargs):

        from authentic2_journal.models import EventKind, KINDS

        if not router.allow_migrate(using, EventKind):

            return

        if EventKind.objects.filter(name__in=KINDS).count() == 9:

            return

        for kind in KINDS:

            EventKind.objects.get_or_create(name=kind)

# authentic - Versatile identity management server

# Copyright (C) 2018 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_journal.models import Line

from authentic2_journal.utils import new_line, new_reference, log_event

from django.utils.translation import ugettext_lazy as _

def sso(user, service):

    representation = _('SSO: user %s on service %s') % (user.email, service)

    log_event(user, service, 'action_sso', representation)

def login(user, provider):

    representation = _('LOGIN: user %s on provider %s') % (user.email, provider)

    log_event(user, provider, 'action_login', representation)

def logout(user, provider):

    representation = _('LOGOUT: user %s on provider %s') % (user.email, provider)

    log_event(user, provider, 'action_logout', representation)

def register(user, ou):

    representation = _('REGISTER: user %s in OU %s') % (user.email, ou)

    log_event(user, ou, 'action_register', representation)

def create(user, obj):

    representation = _('CREATE: user %s on %s') % (user.email, obj)

    log_event(user, obj, 'modification_create', representation)

def update(user, obj):

    representation = _('UPDATE: user %s on %s') % (user.email, obj)

    log_event(user, obj, 'modification_update', representation)

def modify_members(user, role):

    representation = _('MODIFY MEMBERS: user %s on role %s') % (user.email, role.slug)

    log_event(user, role, 'modification_modify_members', representation)

def delete(user, obj):

    representation = _('DELETE: user %s on %s') % (user.email, obj)

    log_event(user, obj, 'modification_delete', representation)

# authentic - Versatile identity management server

# Copyright (C) 2018 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 __future__ import unicode_literals

from django.db import models

try:

    from django.contrib.contenttypes.fields import GenericForeignKey

except ImportError:

    from django.contrib.contenttypes.generic import GenericForeignKey

try:

    from django.contrib.postgres.fields import JSONField # dj1.11 native JSON

except:

    from jsonfield import JSONField # django-jsonfield outer module

from django.contrib.contenttypes.models import ContentType

from django.utils.translation import ugettext_lazy as _

KINDS = (

    'action_register',

    'action_login',

    'action_sso',

    'action_logout',

    'modification_create',

    'modification_update',

    'modification_modify_members',

    'modification_delete',

    'generic_event'

)

def get_choices():

    return tuple((kind, kind) for kind in KINDS)

class EventKind(models.Model):

    name = models.CharField(

            verbose_name=_('name'), choices=get_choices(), max_length=256)

class Line(models.Model):

    timestamp = models.DateTimeField(auto_now=True, verbose_name=_('timestamp'))

    kind = models.ForeignKey(EventKind)

    extra = JSONField(blank=True,null=True, verbose_name=_('content'))

class Reference(models.Model):

    timestamp = models.DateTimeField(verbose_name=_('timestamp'))

    line = models.ForeignKey(Line)

    representation = models.TextField(max_length=256)

    target_ct = models.ForeignKey(ContentType)

    target_id = models.IntegerField(default=0)

    target = GenericForeignKey('target_ct', 'target_id')

    def __init__(self, *args, **kwargs):

        line = kwargs.get('line')

        if line:

            kwargs.update({'timestamp': line.timestamp})

        models.Model.__init__(self, *args, **kwargs)

# authentic - Versatile identity management server

# Copyright (C) 2018 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 django_tables2 as tables

from authentic2_journal.models import Reference

from django.utils.translation import ugettext_lazy as _

class UserEventsTable(tables.Table):

    class Meta:

        attrs = {'class': 'main'}

        model = Reference

        exclude = ['line', 'target_id', 'target_ct']

        empty_text = _('None')

{% extends "authentic2/manager/base.html" %}

{% load i18n staticfiles django_tables2 %}

{% block page_title %}

{% trans "User events journal" %}

{% endblock %}

{% block content %}

  {% render_table table "authentic2/manager/table.html" %}

{% endblock %}

# authentic - Versatile identity management server

# Copyright (C) 2018 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.conf.urls import url

from . import views

urlpatterns = [

    url('^users/(?P<pk>\d+)/journal/$',

        views.user_journal,

        name='a2-journal-user'),

]

# authentic - Versatile identity management server

# Copyright (C) 2018 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 logging

from authentic2_journal.models import EventKind, Line, Reference

from django.contrib.contenttypes.models import ContentType

from django.core.exceptions import ObjectDoesNotExist

def new_line(user, obj, kind='generic_event', extra={}):

    logger = logging.getLogger(__name__)

    try:

        kind = EventKind.objects.get(name=kind)

    except:

        logger.error('EventKind retrieval from model failed for event name %s' %

            kind)

    else:

        try:

            line = Line.objects.create(

                kind=kind,

                extra=extra

            )

        except:

            logger.error("Couldn't create event journal entry for user %s on "

                "object %s" % (user, obj))

        else:

            return line

def new_reference(line, obj, representation):

    logger = logging.getLogger(__name__)

    try:

        ct = ContentType.objects.get_for_model(obj)

    except ObjectDoesNotExist:

        logger.error('ContentType retrieval from model failed for object %s' %

            obj)

        logger.warning('No reference will be created for line %s' % line)

    else:

        try:

            reference = Reference.objects.create(

                    line=line,

                    representation=representation,

                    target_ct=ct,

                    target_id=obj.pk)

        except:

            logger.error("Couldn't create event reference for line %s and "

                "object %s" % (line, obj))

        else:

            return reference

def log_event(user, obj, kind, representation):

    extra = {

        'user_email': user.email,

        'user_pk': user.pk,

        'obj_representation': '%s' % obj,

        'kind': kind

    }

    line = new_line(user, obj, kind, extra)

    new_reference(line, user, representation)

    new_reference(line, obj, representation)

# authentic - Versatile identity management server

# Copyright (C) 2018 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.manager.views import SimpleSubTableView

from authentic2_journal.tables import UserEventsTable

from authentic2_journal.models import Reference

from django.contrib.auth import get_user_model

from django.contrib.contenttypes.models import ContentType

from django.db import models

class BaseJournal(SimpleSubTableView):

    model = models.Model

    table_class = None # TODO generic table class

    template_name = '' # TODO generic template name

    permissions = [] # TODO generic custom_user.view_object permission ?

    filter_table_by_perm = False

    def get_table_queryset(self):

        return Reference.objects.filter(

                target_id=self.object.pk,

                target_ct=ContentType.objects.get_for_model(self.object))

class UserJournal(BaseJournal):

    model = get_user_model()

    table_class = UserEventsTable

    template_name = 'authentic2_journal/user_events.html'

    permissions = ['custom_user.view_user']

    filter_table_by_perm = False

user_journal = UserJournal.as_view()

# -*- coding: utf-8 -*-

from authentic2.a2_rbac.models import Role

from authentic2.a2_rbac.utils import get_default_ou

from authentic2.models import Service

from authentic2.saml.models import (LibertyServiceProvider as SP,

    LibertyProvider as IdP)

from authentic2_journal.event_logger import (sso, login, logout, register,

        create, update, modify_members, delete)

from authentic2_journal.models import Line, Reference

from django.contrib.auth import get_user_model

from django.contrib.contenttypes.models import ContentType

from django.db import transaction

from django_rbac.utils import get_ou_model, get_role_model

def test_action_events(db, app, admin, simple_user, ou1):

    s1 = Service.objects.create(name='s1', slug='s1')

    idp1 = IdP.objects.create(

            name='idp',

            slug='idp',

            protocol_conformance=3)

    sp1 = SP.objects.create(liberty_provider=idp1, enabled=True)

    user_ct = ContentType.objects.get_for_model(get_user_model())

    ou_ct = ContentType.objects.get_for_model(get_ou_model())

    idp_ct = ContentType.objects.get_for_model(IdP)

    sp_ct = ContentType.objects.get_for_model(SP)

    assert not len(Line.objects.all())

    assert not len(Reference.objects.all())

    register(admin, ou1)

    assert len(Line.objects.all()) == 1

    assert len(Reference.objects.filter(

            target_id=admin.pk, target_ct=user_ct)) == 1

    assert len(Reference.objects.filter(

            target_id=ou1.pk, target_ct=ou_ct)) == 1

    sso(admin, sp1)

    assert len(Line.objects.all()) == 2

    assert len(Reference.objects.filter(

            target_id=admin.pk, target_ct=user_ct)) == 2

    assert len(Reference.objects.filter(

            target_id=sp1.pk, target_ct=sp_ct)) == 1

    login(admin, idp1)

    assert len(Line.objects.all()) == 3

    assert len(Reference.objects.filter(

            target_id=admin.pk, target_ct=user_ct)) == 3

    assert len(Reference.objects.filter(

            target_id=idp1.pk, target_ct=idp_ct)) == 1

    logout(admin, idp1)

    assert len(Line.objects.all()) == 4

    assert len(Reference.objects.filter(

            target_id=admin.pk, target_ct=user_ct)) == 4

    assert len(Reference.objects.filter(

            target_id=idp1.pk, target_ct=idp_ct)) == 2

def test_modification_events(db, admin, simple_user):

    r1 = Role.objects.create(name='r1', slug='r1')

    user_ct = ContentType.objects.get_for_model(get_user_model())

    ou_ct = ContentType.objects.get_for_model(get_ou_model())

    role_ct = ContentType.objects.get_for_model(get_role_model())

    assert not len(Line.objects.all())

    assert not len(Reference.objects.all())

    create(admin, simple_user)

    assert len(Line.objects.all()) == 1

    assert len(Reference.objects.filter(

            target_id=admin.pk, target_ct=user_ct)) == 1

    assert len(Reference.objects.filter(

            target_id=simple_user.pk, target_ct=user_ct)) == 1

    update(admin, simple_user)

    assert len(Line.objects.all()) == 2

    assert len(Reference.objects.filter(

            target_id=admin.pk, target_ct=user_ct)) == 2

    assert len(Reference.objects.filter(

            target_id=simple_user.pk, target_ct=user_ct)) == 2

    delete(admin, simple_user)

    assert len(Line.objects.all()) == 3

    assert len(Reference.objects.filter(

            target_id=admin.pk, target_ct=user_ct)) == 3

    assert len(Reference.objects.filter(

            target_id=simple_user.pk, target_ct=user_ct)) == 3

    modify_members(admin, r1)

    assert len(Line.objects.all()) == 4

    assert len(Reference.objects.filter(

            target_id=admin.pk, target_ct=user_ct)) == 4

    assert len(Reference.objects.filter(

            target_id=r1.pk, target_ct=role_ct)) == 1

def test_entries_timestamp_consistency(db, admin):

    r1 = Role.objects.create(name='r1', slug='r1')

    modify_members(admin, r1)

    line = Line.objects.first()

    for ref in Reference.objects.all():

        assert ref.timestamp == line.timestamp

def test_line_retrieval_on_object_deletion(db):

    email = 'toto@nowhere.null'

    role_name = role_slug = 'r1'

    entries_role = []

    entries_user = []

    for line in Line.objects.all():

        if role_name in line.extra['obj_representation']:

            entries_role.append(line)

        if email == line.extra['user_email']:

            entries_user.append(line)

    assert not len(entries_role)

    assert not len(entries_user)

    role = Role.objects.create(name=role_name, slug=role_slug)

    user = get_user_model().objects.create(email=email)

    modify_members(user, role)

    user.delete()

    role.delete()

    entries_role = []

    entries_user = []

    for line in Line.objects.all():

        if role_name in line.extra['obj_representation']:

            entries_role.append(line)

        if email == line.extra['user_email']:

            entries_user.append(line)

    assert len(entries_role)

    assert len(entries_user)

def test_reference_retrieval_after_objects_deletion(db):

    email = 'toto@nowhere.null'

    role_name = role_slug = 'r1'

    entries_role = []

    entries_user = []

    for ref in Reference.objects.all():

        if role_name in ref.representation:

            entries_role.append(ref)

        if email in ref.representation:

            entries_user.append(ref)

    assert not len(entries_role)

    assert not len(entries_user)

    role = Role.objects.create(name=role_name, slug=role_slug)

    user = get_user_model().objects.create(email=email)

    modify_members(user, role)

    user.delete()

    role.delete()

    entries_role = []

    entries_user = []

    for ref in Reference.objects.all():

        if role_name in ref.representation:

            entries_role.append(ref)

        if email in ref.representation:

            entries_user.append(ref)

    assert len(entries_role)

    assert len(entries_user)