Produits Entr'ouvert » Hobo

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

import datetime

import logging

import requests

from django.conf import settings

from django.contrib.auth import get_user_model

from django.core.management.base import BaseCommand

from mellon.models import UserSAMLIdentifier

from hobo.signature import sign_url

class Command(BaseCommand):

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

        logger = logging.getLogger(__name__)

        User = get_user_model()

        self.verbosity = int(options['verbosity'])

        base_idp_url = settings.KNOWN_SERVICES['authentic'].get('idp', {}).get('url', None)

        if not base_idp_url:

            self.stderr.write('resync_users: no base api url known, required for users resynchronization')

            return

        secret = settings.KNOWN_SERVICES['authentic'].get('other', {}).get('secret', None)

        if not secret:

            self.stderr.write(

                'resync_users: no secret is known for authentic instance %s, yet required for users resynchronization',

                base_idp_url,

            )

            return

        url = base_idp_url + 'api/users/synchronization/'

        # check all existing users

        def chunks(l, n):

            for i in range(0, len(l), n):

                yield l[i : i + n]

        unknown_uuids = []

        for usersamlids in chunks(UserSAMLIdentifier.objects.all(), 100):

            uuids = [x.name_id for x in usersamlids]

            resp = requests.post(sign_url(url, key=secret), json={'known_uuids': uuids})

            unknown_uuids.extend(resp.json().get('unknown_uuids'))

        for usersamlid in UserSAMLIdentifier.objects.filter(name_id__in=unknown_uuids):

            if self.verbosity > 0:

                self.stdout.write('deleted user %s with uuid %s' % (usersamlid.user, usersamlid.name_id))

            usersamlid.user.delete()

        # update recently modified users

        url = settings.A2_SOURCE_API_BASE_URL + 'users/?modified__gt=%s' % (

            datetime.datetime.now() - datetime.timedelta(seconds=120)

        ).strftime('%Y-%m-%dT%H:%M:%S')

        resp = requests.get(sign_url(url, key=secret))

        for user_dict in resp.json()['results']:

            try:

                user = User.objects.get(email=user_dict['email'])

            except User.DoesNotExist:

                continue

            modified = []

            for attr in ['first_name', 'last_name', 'username']:

                if attr in user_dict:

                    modified.append(attr)

                    setattr(user, attr, user_dict.get(attr))

            if modified and self.verbosity > 0:

                self.stdout.write(

                    'modified user %s with new attributes %s'

                    % (user, dict((attr, user_dict.get(attr)) for attr in modified))

                )

            user.save()

KNOWN_SERVICES = {

    'authentic': {'idp': {'url': 'https://authentic.dev.publik.null/'}, 'other': {'secret': 'xyz'}},

}

A2_SOURCE_API_BASE_URL = 'https://authentic.dev.publik.love/api/'

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

import json

import os

import sys

import pytest

from django.contrib.auth import get_user_model

from mellon.models import Issuer, UserSAMLIdentifier

from mock import Mock, call, patch

from requests import Response

from hobo.agent.common.management.commands.hobo_deploy import Command as DeployCommand

from hobo.agent.common.management.commands.hobo_deploy import CommandError, replace_file

from hobo.agent.common.management.commands.resync_users import Command as ResyncUsersCommand

from hobo.agent.hobo.management.commands.hobo_deploy import Command as HoboDeployCommand

from hobo.environment.models import Combo, Hobo, Variable, Wcs

from hobo.multitenant.middleware import TenantNotFound

User = get_user_model()

pytestmark = pytest.mark.django_db

@patch('hobo.agent.common.management.commands.hobo_deploy.TenantMiddleware.get_tenant_by_hostname')

def test_resync_users(mocked_get_tenant_by_hostname, db, tmpdir, settings):

    command = DeployCommand()

    command.deploy_specifics = Mock()

    tenant = Mock()

    tenant.get_directory = Mock(return_value=str(tmpdir))

    combo_base_url = 'https://combo.dev.publik.love/'

    wcs_base_url = 'https://wcs.dev.publik.love/'

    tenant_hobo_json = os.path.join(str(tmpdir), 'hobo.json')

    ENVIRONMENT = {

        'services': [

            {

                'service-id': 'combo',

                'base_url': combo_base_url,

                'id': 1,

            },

            {

                'service-id': 'wcs',

                'base_url': wcs_base_url,

                'id': 1,

            },

        ]

    }

    command.deploy_specifics.reset_mock()

    mocked_get_tenant_by_hostname.reset_mock()

    mocked_get_tenant_by_hostname.side_effect = [TenantNotFound, tenant]

    with patch('hobo.agent.common.management.commands.hobo_deploy.call_command') as mocked_call_command:

        command.deploy(combo_base_url, ENVIRONMENT, None)

    assert mocked_call_command.mock_calls == [call('create_tenant', 'combo.dev.publik.love')]

    issuer = Issuer.objects.create(entity_id='My Service', slug='my-service')

    for user_data in [

        ('john', 'doe', 'john.doe', '123'),

        ('will', 'smith', 'will.smith', '456'),

        ('emily', 'joe', 'emily.joe', '789'),

        ('alan', 'brown', 'alan.brown', '012'),

        ('bob', 'miller', 'bob.miller', '345'),

        ('helen', 'jones', 'helen.jones', '678'),

    ]:

        user = User.objects.create(

            first_name=user_data[0],

            last_name=user_data[1],

            username=user_data[2],

            email='%s@nowhere.null' % user_data[2],

        )

        UserSAMLIdentifier.objects.create(user=user, issuer=issuer, name_id=user_data[3])

        user.save()

    users = User.objects.all()

    assert len(users) == 6

    with patch('hobo.agent.common.management.commands.resync_users.requests.post') as mocked_post:

        mocked_post.return_value.ok = True

        mocked_post.return_value.json.return_value = {'unknown_uuids': ['456', '789']}

        with patch('hobo.agent.common.management.commands.resync_users.requests.get') as mocked_get:

            mocked_get.return_value.ok = True

            mocked_get.return_value.json.return_value = {

                'results': [

                    {

                        'email': 'john.doe@nowhere.null',

                        'first_name': 'johnny',

                    }

                ]

            }

            resync_users_command = ResyncUsersCommand()

            resync_users_command.handle(verbosity=1)

    users = User.objects.all()

    assert len(users) == 4

    assert users.get(email='john.doe@nowhere.null').first_name == 'johnny'