Produits Entr'ouvert » Hobo

    REST_FRAMEWORK['DEFAULT_AUTHENTICATION_CLASSES'] = (

        'hobo.rest_authentication.PublikAuthentication',

        'hobo.rest_authentication.APIAccessAuthentication',

    )

# hourly

unique-cron = 1 -1 -1 -1 -1 /usr/bin/hobo-manage tenant_command provision_apiaccess --all-tenants

# Generated by Django 2.2.26 on 2022-06-01 10:19

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [

        ('common', '0004_alter_role_uuid'),

    ]

    operations = [

        migrations.CreateModel(

            name='APIAccess',

            fields=[

                (

                    'id',

                    models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),

                ),

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

                ('identifier', models.CharField(max_length=128, unique=True, verbose_name='identifier')),

                ('key', models.CharField(max_length=128, verbose_name='key')),

                ('description', models.CharField(blank=True, max_length=128, verbose_name='description')),

            ],

        ),

    ]

from django.utils.translation import ugettext_lazy as _

class APIAccess(models.Model):

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

    identifier = models.CharField(max_length=128, unique=True, verbose_name=_('identifier'))

    key = models.CharField(max_length=128, verbose_name=_('key'))

    description = models.CharField(max_length=128, blank=True, verbose_name=_('description'))

# hobo - portal to configure and deploy applications

# Copyright (C) 2015-2022 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 uuid

from django.core.management.base import BaseCommand, CommandError

from hobo.agent.common.models import APIAccess

from hobo.apiaccess.provisionning import notify_agents

class Command(BaseCommand):

    help = 'Add an API Access entry'

    def add_arguments(self, parser):

        parser.add_argument('name', type=str)

    def handle(self, *args, **options):

        name = options['name']

        try:

            apiaccess = APIAccess.objects.get(name=name)

            self.stdout.write(self.style.WARNING('An API Access entry with this name already exists.'))

        except APIAccess.DoesNotExist:

            apiaccess = APIAccess.objects.create(

                identifier=str(uuid.uuid4()), name=name, key=str(uuid.uuid4())

            )

            notify_agents([apiaccess])

            self.stdout.write(self.style.SUCCESS('API Access created.'))

            self.stdout.write(self.style.SUCCESS('Identifier: %s' % apiaccess.identifier))

            self.stdout.write(self.style.SUCCESS('Key: %s' % apiaccess.key))

# hobo - portal to configure and deploy applications

# Copyright (C) 2015-2022 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 uuid

from django.core.management.base import BaseCommand, CommandError

from hobo.agent.common.models import APIAccess

from hobo.apiaccess.provisionning import notify_agents

class Command(BaseCommand):

    help = 'Provision api access'

    def handle(self, *args, **options):

        notify_agents((apiaccess for apiaccess in APIAccess.objects.all()), full=True)

# hobo - portal to configure and deploy applications

# Copyright (C) 2015-2022 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 requests

from django.conf import settings

from hobo.environment.utils import get_installed_services

from hobo.signature import sign_url

def notify_agents(apiaccesses, mode='provision', full=False, sync=False):

    data = {

        '@type': mode,

        'full': full,

        'objects': {

            '@type': 'apiaccess',

            'data': [

                {

                    'identifier': apiaccess.identifier,

                    'key': apiaccess.key,

                    'name': apiaccess.name,

                    'description': apiaccess.description,

                }

                for apiaccess in apiaccesses

            ],

        },

    }

    known_services = getattr(settings, 'KNOWN_SERVICES', {})

    for service_id, services in known_services.items():

        if service_id in ('hobo', 'wcs'):

            continue

        for service in services.values():

            if service['secondary']:

                continue

            provisionning_url = service.get('provisionning-url')

            if not provisionning_url:

                continue

            url = service['provisionning-url'] + '?orig=%s' % service['orig']

            if sync:

                url += '&sync=1'

            try:

                response = requests.put(sign_url(url, service['secret']), json=data)

                response.raise_for_status()

            except requests.RequestException as e:

                pass

from hobo.agent.common.models import APIAccess, Role

    @classmethod

    def provision_apiaccess(cls, issuer, action, data, full=False):

        if action == 'provision':

            identifiers = []

            for record in data:

                try:

                    apiaccess = APIAccess.objects.get(identifier=record['identifier'])

                    apiaccess.name = record['name']

                    apiaccess.key = record['key']

                    apiaccess.description = record['description']

                    apiaccess.save()

                except APIAccess.DoesNotExist:

                    APIAccess.objects.create(

                        identifier=record['identifier'],

                        key=record['key'],

                        name=record['name'],

                        description=record['description'],

                    )

                identifiers.append(record['identifier'])

            if full:

                APIAccess.objects.exclude(identifier__in=identifiers).delete()

        if action == 'deprovision':

            for record in data:

                try:

                    apiacess = APIAccess.objects.get(key=record['key']).delete()

                except APIAccess.DoesNotExist:

                    pass

from hobo.agent.common.models import APIAccess

class APIAccessAuthenticationFailed(exceptions.APIException):

    status_code = status.HTTP_401_UNAUTHORIZED

    def __init__(self, err_desc):

        self.detail = {'err': 1, 'err_desc': err_desc}

class APIAccessAuthentication(authentication.BasicAuthentication):

    def authenticate_credentials(self, identifier, key, request=None):

        try:

            APIAccess.objects.get(identifier=identifier, key=key)

        except APIAccess.DoesNotExist:

            raise APIAccessAuthenticationFailed('apiaccess-not-found')

        if hasattr(settings, 'HOBO_ANONYMOUS_SERVICE_USER_CLASS'):

            klass = import_string(settings.HOBO_ANONYMOUS_SERVICE_USER_CLASS)

            return (klass(), None)

        raise APIAccessAuthenticationFailed('apiaccess-configuration-error')

    'hobo.apiaccess',

import base64

from io import StringIO

import mock

import pytest

import responses

from django.core.management import call_command

from django.test import RequestFactory

import hobo.apiaccess.management.commands.add_apiaccess

from hobo.agent.common.models import APIAccess

from hobo.apiaccess.provisionning import notify_agents

from hobo.rest_authentication import APIAccessAuthentication, APIAccessAuthenticationFailed

from hobo.signature import sign_url

def test_add_apiaccess_command(db, monkeypatch):

    assert APIAccess.objects.count() == 0

    notify_agents = mock.Mock()

    monkeypatch.setattr(hobo.apiaccess.management.commands.add_apiaccess, 'notify_agents', notify_agents)

    out = StringIO()

    call_command('add_apiaccess', 'foo', stdout=out)

    assert APIAccess.objects.count() == 1

    apiaccess = APIAccess.objects.get(name='foo')

    notify_agents.assert_called_once_with([apiaccess])

    stdout = out.getvalue()

    assert 'API Access created.' in stdout

    assert 'Identifier: %s' % apiaccess.identifier in stdout

    assert 'Key: %s' % apiaccess.key in stdout

    notify_agents.reset_mock()

    out = StringIO()

    call_command('add_apiaccess', 'foo', stdout=out)

    assert APIAccess.objects.count() == 1

    notify_agents.assert_not_called()

    assert 'An API Access entry with this name already exists.' in out.getvalue()

def test_apiaccess_authentication(db, settings):

    settings.HOBO_ANONYMOUS_SERVICE_USER_CLASS = 'hobo.rest_authentication.AnonymousAdminServiceUser'

    URL = '/api/'

    auth_headers = {

        'HTTP_AUTHORIZATION': 'Basic ' + base64.b64encode('foo:bar'.encode('utf-8')).decode('utf-8'),

    }

    factory = RequestFactory()

    request = factory.get('/api/', **auth_headers)

    apiaccess_authentication = APIAccessAuthentication()

    with pytest.raises(APIAccessAuthenticationFailed) as excinfo:

        apiaccess_authentication.authenticate(request)

    assert 'apiaccess-not-found' in str(excinfo.value)

    apiaccess = APIAccess.objects.create(identifier='foo', key='bar')

    result = apiaccess_authentication.authenticate(request)

    assert isinstance(result, tuple)

    assert len(result) == 2

    assert result[1] is None

    user = result[0]

    assert user.is_staff

    assert user.is_anonymous

    assert user.is_authenticated

def test_apiaccess_notify_agents(db, settings):

    settings.KNOWN_SERVICES = {

        'chrono': {

            'foo': {

                'title': 'Foo',

                'url': 'https://chrono.example.invalid/',

                'orig': 'example.org',

                'secret': 'xxx',

                'provisionning-url': 'https://chrono.example.invalid/__provision__/',

                'secondary': False,

            }

        },

        'combo': {

            'bar': {

                'title': 'Bar',

                'url': 'https://combo.example.invalid/',

                'orig': 'example.org',

                'secret': 'xxx',

                'provisionning-url': 'https://combo.example.invalid/__provision__/',

                'secondary': False,

            }

        },

    }

    with responses.RequestsMock() as rsps:

        rsps.put(

            'https://chrono.example.invalid/__provision__/',

            status=200,

            match=[responses.matchers.query_param_matcher({}, strict_match=False)],

        )

        rsps.put(

            'https://combo.example.invalid/__provision__/',

            status=200,

            match=[responses.matchers.query_param_matcher({}, strict_match=False)],

        )

        apiaccess = APIAccess.objects.create(identifier='foo', key='bar', name='Foo')

        notify_agents([apiaccess])

        assert len(rsps.calls) == 2

def test_apiaccess_provision(db, app, settings):

    settings.HOBO_ANONYMOUS_SERVICE_USER_CLASS = 'hobo.rest_authentication.AnonymousAdminServiceUser'

    settings.KNOWN_SERVICES = {

        'chrono': {

            'foo': {

                'title': 'Foo',

                'url': 'https://chrono.example.invalid/',

                'verif_orig': 'chrono.example.invalid',

                'secret': 'xxx',

                'provisionning-url': 'https://chrono.example.invalid/__provision__/',

                'secondary': False,

            }

        },

        'authentic': {

            'bar': {

                'title': 'Bar',

                'url': 'https://authentic.example.invalid/',

                'verif_orig': 'authentic.example.invalid',

                'secret': 'xxx',

                'provisionning-url': 'https://authentic.example.invalid/__provision__/',

                'secondary': False,

            }

        },

        'hobo': {

            'hobo': {

                'title': 'Hobo',

                'url': 'https://hobo.example.invalid/',

                'verif_orig': 'hobo.example.invalid',

                'secret': 'xxx',

                'provisionning-url': 'https://hobo.example.invalid/__provision__/',

                'secondary': False,

            }

        },

    }

    # create one api access

    notification = {

        '@type': 'provision',

        'objects': {

            '@type': 'apiaccess',

            'data': [{'key': 'key', 'identifier': 'id', 'name': 'name', 'description': 'description'}],

        },

    }

    assert APIAccess.objects.count() == 0

    resp = app.put_json(sign_url('/__provision__/?orig=%s' % 'hobo.example.invalid', 'xxx'), notification)

    assert APIAccess.objects.count() == 1

    apiaccess = APIAccess.objects.get(identifier='id')

    assert apiaccess.key == 'key'

    assert apiaccess.name == 'name'

    assert apiaccess.description == 'description'

    # update the api access

    notification = {

        '@type': 'provision',

        'objects': {

            '@type': 'apiaccess',

            'data': [

                {'key': 'new-key', 'identifier': 'id', 'name': 'new-name', 'description': 'new-description'}

            ],

        },

    }

    resp = app.put_json(sign_url('/__provision__/?orig=%s' % 'hobo.example.invalid', 'xxx'), notification)

    assert APIAccess.objects.count() == 1

    apiaccess = APIAccess.objects.get(identifier='id')

    assert apiaccess.key == 'new-key'

    assert apiaccess.name == 'new-name'

    assert apiaccess.description == 'new-description'

    # delete the api access

    notification = {

        '@type': 'deprovision',

        'objects': {

            '@type': 'apiaccess',

            'data': [

                {'key': 'new-key', 'identifier': 'id', 'name': 'new-name', 'description': 'new-description'}

            ],

        },

    }

    resp = app.put_json(sign_url('/__provision__/?orig=%s' % 'hobo.example.invalid', 'xxx'), notification)

    assert APIAccess.objects.count() == 0

    # create three api access

    notification = {

        '@type': 'provision',

        'objects': {

            '@type': 'apiaccess',

            'data': [

                {'key': 'key1', 'identifier': 'id1', 'name': 'name1', 'description': 'description1'},

                {'key': 'key2', 'identifier': 'id2', 'name': 'name2', 'description': 'description2'},

                {'key': 'key3', 'identifier': 'id3', 'name': 'name3', 'description': 'description3'},

            ],

        },

    }

    assert APIAccess.objects.count() == 0

    resp = app.put_json(sign_url('/__provision__/?orig=%s' % 'hobo.example.invalid', 'xxx'), notification)

    assert APIAccess.objects.count() == 3

    assert APIAccess.objects.get(identifier='id1')

    assert APIAccess.objects.get(identifier='id2')

    assert APIAccess.objects.get(identifier='id3')

    # full provisionning

    notification = {

        '@type': 'provision',

        'full': True,

        'objects': {

            '@type': 'apiaccess',

            'data': [

                {'key': 'key1', 'identifier': 'id1', 'name': 'name1', 'description': 'description1'},

                {'key': 'key2', 'identifier': 'id2', 'name': 'name2', 'description': 'description2'},

            ],

        },

    }

    resp = app.put_json(sign_url('/__provision__/?orig=%s' % 'hobo.example.invalid', 'xxx'), notification)

    assert APIAccess.objects.count() == 2

    assert APIAccess.objects.get(key='key1')

    assert APIAccess.objects.get(key='key2')

        responses