Produits Entr'ouvert » Authentic 2

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

              'authentic2-journal = authentic2_journal:Plugin',

    A2_JOURNAL_EVENT_LOGGER=Setting(default=None, definition='Event logger class for the journal app.'),

from authentic2.forms.widgets import DateTimeWidget

class EventTimewindowForm(forms.Form):

    after = forms.DateTimeField(label=_('After'), widget=DateTimeWidget)

    before = forms.DateTimeField(label=_('Before'), widget=DateTimeWidget)

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

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

            user_views.user_journal,

            name='a2-manager-user-journal'),

from authentic2_journal.views import ManagerBaseJournal

from authentic2_journal.tables import UserEventsTable

class UserJournal(ManagerBaseJournal):

    model = get_user_model()

    table_class = UserEventsTable

    template_name = 'authentic2/manager/user_journal.html'

    permissions = ['custom_user.view_user']

    filter_table_by_perm = False

user_journal = UserJournal.as_view()

from .views import (logged_in, edit_profile, email_change, email_change_verify, profile,

        user_profile_journal)

    url('^journal/$', user_profile_journal, name='user-profile-journal'),

    'authentic2_journal',

{% extends "authentic2/base-page.html" %}

{% load i18n %}

{% load authentic2_journal %}

{% block content %}

  <h1>{% trans "User journal" %}</h1>

  <ul>

  {% for line in object_list %}

      <li>{{ line.timestamp }} - {{ line|to_representation }}</li>

  {% empty %}

      <li>{% trans "No journal entry yet." %}</li>

  {% endfor %}

  </ul>

  {% block pagination %}

    {% include "gadjo/pagination.html" %}

  {% endblock %}

  <form action="{% url 'user-profile-journal' %}" method="get">

    {% csrf_token %}

    {{ form }}

    <input type="submit" value="Submit">

  </form>

{% endblock %}

from datetime import datetime

from django.views.generic.list import ListView

from django.contrib.auth import REDIRECT_FIELD_NAME, get_user_model

from django.contrib.contenttypes.models import ContentType

from authentic2_journal.models import Reference, Line

class UserProfileJournal(ListView, FormView):

    model = Line

    form_class = forms.EventTimewindowForm

    paginate_by = 10

    def get_queryset(self):

        timewindow = {}

        for qs_param, qs_suffix in {'after': 'gte', 'before': 'lte'}.items():

            delimiter = self.request.GET.get(qs_param)

            if delimiter:

                try:

                    dt = datetime.strptime(delimiter, '%d/%m/%Y %H:%M:%S')

                except:

                    logger.info('UserProfileJournal wrong %s timestamp query '

                            'string parameter: %s' % (qs_param, delimiter))

                else:

                    timewindow.update({'timestamp__%s' % qs_suffix: dt})

        """

        Timestamp consistency allows for time window filtering on Reference

        objects directly

        """

        return [ref.line for ref in Reference.objects.filter(

            target_id=self.request.user.pk,

            target_ct=ContentType.objects.get_for_model(get_user_model()),

            **timewindow)]

    def get_template_names(self):

        return [

            'profiles/user_profile_journal.html',

            'authentic2/user_profile_journal.html',

        ]

user_profile_journal = login_required(UserProfileJournal.as_view())

# 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_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'

# 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.contrib.contenttypes.models import ContentType

from django.template.loader import render_to_string, select_template

_kind_registry = {}

class KindMetaclass(type):

    def __init__(self, name, bases, dct):

        super(KindMetaclass, self).__init__(name, bases, dct)

        if name != "BaseKind":

            assert set(dct.keys()) >= set(['name', 'template_names'])

            assert dct['name'] not in _kind_registry

            _kind_registry[dct['name']] = self

class BaseKind(object):

    __metaclass__ = KindMetaclass

    long_template_names = []

    template_names = []

    def format(self, line):

        return select_template(self.template_names).render(context=line.extra_json)

    def long_format(self, line):

        if not self.long_template_names:

            return ''

        return select_template(self.long_template_names).render(context=line.extra_json)

    def log_real(self, **kwargs):

        from authentic2_journal.models import Line, get_kind

        kind = get_kind(self.name)

        if isinstance(kind, tuple):

            kind = kind[0]

        return Line.objects.create(kind=kind, extra_json=kwargs)

    def create_reference(self, line, obj, **kwargs):

        from authentic2_journal.models import Reference

        reference = Reference.objects.create(

                line=line,

                target_ct=ContentType.objects.get_for_model(obj),

                target_id=obj.pk)

    def log(self, user, obj):

        line = self.log_real(user_pk=user.pk, obj_pk=obj.pk, user='%s' % user, obj='%s' % obj)

        if line:

            self.create_reference(line, user)

            self.create_reference(line, obj)

class SSOKind(BaseKind):

    name = 'sso'

    template_names = ['authentic2_journal/sso.html']

class LoginKind(BaseKind):

    name = 'login'

    template_names = ['authentic2_journal/login.html']

class LogoutKind(BaseKind):

    name = 'logout'

    template_names = ['authentic2_journal/logout.html']

class RegisterKind(BaseKind):

    name = 'register'

    template_names = ['authentic2_journal/register.html']

class CreateKind(BaseKind):

    name = 'create'

    template_names = ['authentic2_journal/create.html']

class UpdateKind(BaseKind):

    name = 'update'

    template_names = ['authentic2_journal/update.html']

class ModifyMembersKind(BaseKind):

    name = 'modify_members'

    template_names = ['authentic2_journal/modify_members.html']

class DeleteKind(BaseKind):

    name = 'delete'

    template_names = ['authentic2_journal/delete.html']

from authentic2_journal import kinds, models

class Logger(object):

    def __init__(self, kind):

        self.kind = kinds._kind_registry[kind]

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

        kind_model = models.get_kind(self.kind.name)

        if isinstance(kind_model, tuple):

            kind_model = kind_model[0]

        kind = kind_model.concrete()

        return self.kind.log(kind, *args, **kwargs)

class LoggerHub(object):

    def __getattr__(self, kind_name):

        return Logger(kind_name)

logger = LoggerHub()

# 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 authentic2.decorators import GlobalCache

from authentic2_journal import kinds

from django.db import models

try:

    from django.contrib.contenttypes.fields import GenericForeignKey

except ImportError:

    from django.contrib.contenttypes.generic import GenericForeignKey

from jsonfield import JSONField # django-jsonfield outer module

from django.contrib.contenttypes.models import ContentType

from django.utils.translation import ugettext_lazy as _

class Kind(models.Model):

    name = models.CharField(max_length=128, null=True, verbose_name='name')

    @property

    def concrete(self):

        return kinds._kind_registry[self.name]

# get_kind() should be cached, like ContentType are, because they never vary

@GlobalCache

def get_kind(name):

    return Kind.objects.get_or_create(name=name)

class Line(models.Model):

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

    kind = models.ForeignKey(Kind)

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

    def format(self):

        return self.kind.concrete().format(self)

    def long_format(self):

        return self.kind.concrete().long_format(self)

class Reference(models.Model):

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

    line = models.ForeignKey(Line)

    extra_json = JSONField(blank=False, null=False, verbose_name=_('additional description'))

    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,

                'extra_json': line.extra_json

            })

        return super(Reference, self).__init__(*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 Line

from django.utils.translation import ugettext_lazy as _

class UserEventsTable(tables.Table):

    def render_kind(self, value):

        return value.name

    def render_extra_json(self, value, record):

        return record.format() or value

    class Meta:

        attrs = {'class': 'main'}

        model = Line

        order_by = ('-timestamp', 'id')

        per_page = 10

        empty_text = _('No journal entry yet.')

{% load i18n %}

{% blocktrans %}{{user}} performed create on {{obj}}{% endblocktrans %}

{% load i18n %}

{% blocktrans %}{{user}} performed delete on {{obj}}{% endblocktrans %}

{% load i18n %}

{% blocktrans %}{{user}} performed login on {{provider}}{% endblocktrans %}

{% load i18n %}

{% blocktrans %}{{user}} performed logout on {{provider}}{% endblocktrans %}

{% load i18n %}

{% blocktrans %}{{user}} performed modify members on {{role}}{% endblocktrans %}

{% load i18n %}

{% blocktrans %}{{user}} registered on {{ou}}{% endblocktrans %}

{% load i18n %}

{% blocktrans %}{{user}} performed sso on {{service}}{% endblocktrans %}

{% load i18n %}

{% blocktrans %}{{user}} performed update on {{obj}}{% endblocktrans %}

# 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 import template

register = template.Library()

@register.filter

def to_representation(line):

    return line.format()

@register.filter

def to_long_representation(line):

    return line.long_format()

# 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.models import Reference

from django.contrib.auth import get_user_model

from django.contrib.contenttypes.models import ContentType

class ManagerBaseJournal(SimpleSubTableView):

    model = None

    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

    table_pagination = True

    def get_table_queryset(self):

        return [ref.line for ref in Reference.objects.filter(

                target_id=self.object.pk,

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

@pytest.fixture

def event_logger(db):

    from authentic2_journal.logger import LoggerHub

    return LoggerHub()

# -*- 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.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

from utils import login

def test_action_events(db, event_logger, 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())

    event_logger.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

    event_logger.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

    event_logger.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

    event_logger.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, event_logger, 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())

    event_logger.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

    event_logger.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

    event_logger.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

    event_logger.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, event_logger, admin):

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

    event_logger.modify_members(admin, r1)

    line = Line.objects.first()

    for ref in Reference.objects.all():

        assert ref.timestamp == line.timestamp

def test_line_retrieval_after_object_deletion(db, event_logger):

    email = u'toto@nowhere.null'

    role_name = role_slug = u'r1'

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

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

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

    event_logger.modify_members(user, role)

    user.delete()

    role.delete()

    entries_role = []

    entries_user = []

    for line in Line.objects.all():

        if role_name in line.extra_json['obj']:

            entries_role.append(line)

        if line.extra_json['user'].startswith(email):

            entries_user.append(line)

    assert len(entries_role)

    assert len(entries_user)

def test_reference_retrieval_after_objects_deletion(db, event_logger):

    email = 'toto@nowhere.null'

    role_name = role_slug = 'r1'

    assert not(Reference.objects.all())

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

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

    event_logger.modify_members(user, role)

    user.delete()

    role.delete()

    entries_role = []

    entries_user = []

    for ref in Reference.objects.all():

        if role_name in ref.extra_json['obj']:

            entries_role.append(ref)

        if ref.extra_json.get('user', '').startswith(email):

            entries_user.append(ref)

    assert len(entries_role)

    assert len(entries_user)

def test_backoffice_user_journal_view(db, event_logger, app, admin, simple_user, ou1):

    login(app, admin, '/manage/')

    url = u'/manage/users/%s/journal/' % admin.pk

    event_logger.create(admin, simple_user)

    event_logger.create(admin, ou1)

    response = app.get(url, status=200)

    table_lines = response.html.find(

            'div', attrs={'class': 'table-container'}).find(

            'table', 'main').find('tbody').find_all('tr')

    assert len(table_lines) == 2

    # assumptions on table content

    for line in table_lines:

        assert 'create' in line.text

        assert '%s' % admin in line.text

        assert '%s' % simple_user in line.text or '%s' % ou1 in line.text

def test_accounts_user_journal_view(db, event_logger, app, admin, simple_user, ou1):

    login(app, admin, '/accounts/')

    url = u'/accounts/journal/'

    event_logger.create(admin, simple_user)

    event_logger.create(admin, ou1)

    response = app.get(url, status=200)

    ref_list = response.html.find('div', attrs={'id': 'content'}).find('ul')

    assert len(ref_list.find_all('li')) == 2

    # assumptions on table content

    for entry in ref_list.find_all('li'):

        assert 'create' in entry.text

        assert '%s' % admin in entry.text

        assert '%s' % simple_user in entry.text or '%s' % ou1 in entry.text

def test_backoffice_unauthorized_access(db, event_logger, app, admin, simple_user, ou1):

    event_logger.create(admin, ou1)

    event_logger.update(admin, ou1)

    login(app, simple_user, '/')

    url = u'/manage/users/%s/journal/' % admin.pk

    response = app.get(url, status=403)

def test_log_event_escalate_error(db, event_logger, app, admin, simple_user, ou1):

    try:

        event_logger.spoke_to(admin, simple_user)

    except:

        pass

    else:

        assert 0