Produits Entr'ouvert » Hobo

import logging

from rest_framework import authentication, exceptions

from hobo import signature

from authentic2.saml import models as saml_models

from django.contrib.auth import get_user_model

from django.conf import settings

from django.contrib.auth.models import AnonymousUser

class AnonymousServiceUser(AnonymousUser):

    '''This virtual user hold permissions for other publik services'''

    def is_authenticated(self):

        return True

    def has_perm(self, perm_or_perms, obj=None):

        return 'a2_rbac.change_role' == perm_or_perms

    def __unicode__(self):

        return 'Publik Authentikation Service User'

class PublikAuthentication(authentication.BaseAuthentication):

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

        self.logger = logging.getLogger(__name__)

        super(PublikAuthentication, self).__init__(*args, **kwargs)

    def resolve_user(self, request):

        if 'NameID' in request.GET:

            name_id = request.GET['NameID']

            User = get_user_model()

            try:

                return User.objects.get(uuid=name_id)

            except User.DoesNotExist:

                self.logger.warning('no user found %r', uuid)

                raise exceptions.AuthenticationFailed('No user matches uuid=%r' % uuid)

        else:

            self.logger.info('anonymous signature validated')

            return AnonymousServiceUser()

    def get_orig_key(self, orig):

        if not hasattr(settings, 'KNOWN_SERVICES'):

            self.logger.warning('no known services')

            raise exceptions.AuthenticationFailed('No KNOWN_SERVICES setting')

        for service_id in settings.KNOWN_SERVICES:

            for slug, service in settings.KNOWN_SERVICES[service_id].iteritems():

                if service.get('verif_orig') == orig and service.get('secret'):

                    return service['secret']

        self.logger.warning('no secret found for origin %r', orig)

        raise exceptions.AuthenticationFailed('no secret found for origin %r' % orig)

    def authenticate(self, request):

        full_path = request.get_full_path()

        if not request.GET.get('orig') or not request.GET.get('signature'):

            return None

        key = self.get_orig_key(request.GET['orig'])

        if not signature.check_url(full_path, key):

            self.logger.warning('invalid signature')

            raise exceptions.AuthenticationFailed('Invalid signature')

        user = self.resolve_user(request)

        self.logger.info('user authenticated with signature %s', user)

        return (user, None)

                     'service-id': 'authentic',

                     'service-id': 'welco',

                     'secret_key': 'abcdef',

import pytest

import urllib

from rest_framework.exceptions import AuthenticationFailed

from django.contrib.auth import get_user_model

from django.test import RequestFactory

from tenant_schemas.utils import tenant_context

from hobo import signature, rest_authentication

pytestmark = pytest.mark.django_db

def test_publik_authentication(tenant, settings):

    with tenant_context(tenant):

        key = settings.KNOWN_SERVICES['welco']['other']['secret']

        settings.HOBO_ROLE_EXPORT = False

        User = get_user_model()

        user = User.objects.create(username='foo', password='foo')

        ORIG = 'other.example.net'

        AUTH_QUERY = '&NameID=%s&&orig=%s' % (user.uuid, urllib.quote(ORIG))

        URL = '/api/?coucou=zob'

        factory = RequestFactory()

        request = factory.get(signature.sign_url(URL + AUTH_QUERY, key))

        publik_authentication = rest_authentication.PublikAuthentication()

        result = publik_authentication.authenticate(request)

        assert result is not None

        assert isinstance(result, tuple)

        assert len(result) == 2

        assert result[0] == user

        assert result[1] == None

        # Failure

        request = factory.get(signature.sign_url(URL + AUTH_QUERY, key+'zob'))

        publik_authentication = rest_authentication.PublikAuthentication()

        with pytest.raises(AuthenticationFailed):

            publik_authentication.authenticate(request)