Produits Entr'ouvert » Hobo

class Plugin:

    def get_before_urls(self):

        from . import urls

        return urls.urlpatterns

    def get_a2_plugin(self):

        return Plugin()

            leftover_audience = self.notify_agents_http(data)

            if not leftover_audience:

                return

            logger.info('leftover AMQP audience: %s', leftover_audience)

            data['audience'] = leftover_audience

    def get_http_services_by_url(self):

        services_by_url = {}

        for services in settings.KNOWN_SERVICES.values():

            for service in services.values():

                if service.get('provisionning-url'):

                    services_by_url[service['saml-sp-metadata-url']] = service

        return services_by_url

    def notify_agents_http(self, data):

        services_by_url = self.get_http_services_by_url()

        audience = data.get('audience')

        rest_audience = [x for x in audience if x in services_by_url]

        leftover_audience = audience

        for audience in rest_audience:

            service = services_by_url[audience]

            data['audience'] = [audience]

            try:

                response = requests.put(

                    sign_url(service['provisionning-url'] + '?orig=%s' % service['orig'], service['secret']),

                    json=data)

                response.raise_for_status()

            except requests.RequestException as e:

                logger.error(u'error provisionning to %s (%s)', audience, e)

            else:

                leftover_audience.remove(audience)

        return leftover_audience

# hobo - portal to configure and deploy applications

# Copyright (C) 2015-2021  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(r'^api/provision/$', views.provision_view),

]

# hobo - portal to configure and deploy applications

# Copyright (C) 2015-2021  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 import settings

from rest_framework import permissions, serializers, status

from rest_framework.response import Response

from rest_framework.generics import GenericAPIView

from . import provisionning

class ProvisionSerializer(serializers.Serializer):

    user_uuid = serializers.CharField(required=False)

    role_uuid = serializers.CharField(required=False)

    service_type = serializers.CharField(required=False)

    service_url = serializers.CharField(required=False)

    def validate(self, data):

        if not (data.get('user_uuid') or data.get('role_uuid')):

            raise serializers.ValidationError('must provide user_uuid or role_uuid')

        if data.get('user_uuid') and data.get('role_uuid'):

            raise serializers.ValidationError('cannot provision both user & role')

        return data

class ProvisionView(GenericAPIView):

    permission_classes = (permissions.IsAuthenticated,)

    serializer_class = ProvisionSerializer

    def post(self, request):

        serializer = self.get_serializer(data=request.data)

        if not serializer.is_valid():

            return Response({'err': 1, 'errors': serializer.errors}, status.HTTP_400_BAD_REQUEST)

        engine = ApiProvisionningEngine(

                service_type=serializer.validated_data.get('service_type'),

                service_url=serializer.validated_data.get('service_url'),

                )

        user_uuid = serializer.validated_data.get('user_uuid')

        role_uuid = serializer.validated_data.get('role_uuid')

        if user_uuid:

            try:

                user = provisionning.User.objects.get(uuid=user_uuid)

            except provisionning.User.DoesNotExist:

                return Response({'err': 1, 'err_desc': 'unknown user UUID'})

            engine.notify_users(ous=None, users=[user])

        elif role_uuid:

            try:

                role = provisionning.Role.objects.get(uuid=role_uuid)

            except provisionning.Role.DoesNotExist:

                return Response({'err': 1, 'err_desc': 'unknown role UUID'})

            ous = {ou.id: ou for ou in provisionning.OU.objects.all()}

            engine.notify_roles(ous=ous, roles=[role])

        response = {'err': 0}

        response['leftover_audience'] = engine.leftover_audience

        return Response(response)

provision_view = ProvisionView.as_view()

class ApiProvisionningEngine(provisionning.Provisionning):

    def __init__(self, service_type=None, service_url=None):

        super().__init__()

        self.service_type = service_type

        self.service_url = service_url

    def get_http_services_by_url(self):

        if self.service_type:

            services_by_url = {}

            for service in settings.KNOWN_SERVICES[self.service_type].values():

                if service.get('provisionning-url'):

                    services_by_url[service['saml-sp-metadata-url']] = service

        else:

            services_by_url = super().get_http_services_by_url()

        if self.service_url:

            services_by_url = {k: v for k, v in services_by_url.items() if self.service_url in v['url']}

        return services_by_url

    def notify_agents(self, data):

        self.leftover_audience = self.notify_agents_http(data)

        # only include filtered services in leftovers

        services_by_url = self.get_http_services_by_url()

        self.leftover_audience = [x for x in self.leftover_audience if x in services_by_url]

                        'url': 'http://other.example.net',

                        'base_url': 'http://other.example.net',

                        'provisionning-url': 'http://other.example.net/__provision__/',

                        'saml-sp-metadata-url': 'http://other.example.net/metadata/',

                    {

                        'slug': 'more',

                        'title': 'More',

                        'service-id': 'wcs',

                        'secret_key': 'abcdef',

                        'url': 'http://more.example.net',

                        'base_url': 'http://more.example.net',

                        'provisionning-url': 'http://more.example.net/__provision__/',

                        'saml-sp-metadata-url': 'http://more.example.net/metadata/',

                    },

from hobo import signature

def test_provisionning_api(transactional_db, app_factory, tenant, settings, caplog):

    with tenant_context(tenant):

        # create providers so notification messages have an audience.

        LibertyProvider.objects.create(ou=get_default_ou(), name='provider', slug='provider',

                                       entity_id='http://other.example.net/metadata/',

                                       protocol_conformance=lasso.PROTOCOL_SAML_2_0)

        LibertyProvider.objects.create(ou=get_default_ou(), name='provider2', slug='provider2',

                                       entity_id='http://more.example.net/metadata/',

                                       protocol_conformance=lasso.PROTOCOL_SAML_2_0)

        role = Role.objects.create(name='coin', ou=get_default_ou())

        user = User.objects.create(username='Étienne',

                    email='etienne.dugenou@example.net',

                    first_name='Étienne',

                    last_name='Dugenou',

                    ou=get_default_ou())

    app = app_factory(tenant)

    resp = app.post_json('/api/provision/', {}, status=403)

    orig = settings.KNOWN_SERVICES['welco']['other']['verif_orig']

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

    resp = app.post_json(signature.sign_url('/api/provision/?orig=%s' % orig, key), {}, status=400)

    assert resp.json['errors']['__all__'] == ['must provide user_uuid or role_uuid']

    resp = app.post_json(signature.sign_url('/api/provision/?orig=%s' % orig,

        key), {'user_uuid': 'xxx', 'role_uuid': 'yyy'}, status=400)

    assert resp.json['errors']['__all__'] == ['cannot provision both user & role']

    resp = app.post_json(signature.sign_url('/api/provision/?orig=%s' % orig,

        key), {'user_uuid': 'xxx'}, status=200)

    assert resp.json == {'err': 1, 'err_desc': 'unknown user UUID'}

    resp = app.post_json(signature.sign_url('/api/provision/?orig=%s' % orig,

        key), {'role_uuid': 'xxx'}, status=200)

    assert resp.json == {'err': 1, 'err_desc': 'unknown role UUID'}

    with patch('hobo.agent.authentic2.provisionning.requests.put') as requests_put:

        resp = app.post_json(signature.sign_url('/api/provision/?orig=%s' % orig,

            key), {'user_uuid': user.uuid})

        assert requests_put.call_count == 2

    with patch('hobo.agent.authentic2.provisionning.requests.put') as requests_put:

        resp = app.post_json(signature.sign_url('/api/provision/?orig=%s' % orig,

            key), {'user_uuid': user.uuid, 'service_type': 'welco'})

        assert requests_put.call_count == 1

    with patch('hobo.agent.authentic2.provisionning.requests.put') as requests_put:

        resp = app.post_json(signature.sign_url('/api/provision/?orig=%s' % orig,

            key), {'user_uuid': user.uuid, 'service_url': 'example.net'})

        assert requests_put.call_count == 2

    with patch('hobo.agent.authentic2.provisionning.requests.put') as requests_put:

        resp = app.post_json(signature.sign_url('/api/provision/?orig=%s' % orig,

            key), {'role_uuid': role.uuid})

        assert requests_put.call_count == 2