Produits Entr'ouvert » Hobo

                    provider, service_created = None, False

                    for legacy_urls in service.get('legacy_urls', []):

                        try:

                            provider = LibertyProvider.objects.get(

                                entity_id=legacy_urls['saml-sp-metadata-url'],

                                protocol_conformance=lasso.PROTOCOL_SAML_2_0,

                            )

                            provider.entity_id = sp_url

                            break

                        except LibertyProvider.DoesNotExist:

                            pass

                    if not provider:

                        provider, service_created = LibertyProvider.objects.get_or_create(

                            entity_id=sp_url, protocol_conformance=lasso.PROTOCOL_SAML_2_0

                        )

        legacy_domain = None

        tenant_created = False

            # might be a domain change request

            for legacy_urls in self.me.get('legacy_urls', []):

                old_domain = urlparse.urlparse(legacy_urls['base_url']).netloc.split(':')[0]

                try:

                    tenant = TenantMiddleware.get_tenant_by_hostname(old_domain)

                    legacy_domain = old_domain

                    break

                except TenantNotFound:

                    pass

            call_command('create_tenant', domain, legacy_hostname=legacy_domain)

            tenant_created = True

        if tenant_created and legacy_domain and self.me['service-id'] == 'hobo':

            # need to update local hobo

            with tenant_context(tenant):

                import hobo.environment.utils

                local_hobo = hobo.environment.utils.get_or_create_local_hobo()

                if local_hobo.get_base_url_path() != self.me['base_url']:

                    local_hobo.change_base_url(self.me['base_url'])

                    local_hobo.save()

from hobo.environment.utils import get_or_create_local_hobo

                service, created = None, False

                for legacy_urls in service_dict.get('legacy_urls', []):

                    try:

                        service = service_klass.objects.get(base_url=legacy_urls['base_url'], secondary=True)

                        service.change_base_url(service_dict['base_url'])

                        break

                    except service_klass.DoesNotExist:

                        pass

                if not service:

                    service, created = service_klass.objects.get_or_create(

                        base_url=service_dict['base_url'],

                        secondary=True,

                        defaults={

                            'title': service_dict['title'],

                            'slug': service_slug,

                            'secret_key': service_dict.get('secret_key'),

                        },

                    )

# Generated by Django 2.2.24 on 2021-12-02 13:55

import django.contrib.postgres.fields.jsonb

from django.db import migrations

class Migration(migrations.Migration):

    dependencies = [

        ('environment', '0023_populate_local_hobo'),

    ]

    operations = [

        migrations.AddField(

            model_name='authentic',

            name='legacy_urls',

            field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),

        ),

        migrations.AddField(

            model_name='bijoe',

            name='legacy_urls',

            field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),

        ),

        migrations.AddField(

            model_name='chrono',

            name='legacy_urls',

            field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),

        ),

        migrations.AddField(

            model_name='combo',

            name='legacy_urls',

            field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),

        ),

        migrations.AddField(

            model_name='fargo',

            name='legacy_urls',

            field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),

        ),

        migrations.AddField(

            model_name='hobo',

            name='legacy_urls',

            field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),

        ),

        migrations.AddField(

            model_name='passerelle',

            name='legacy_urls',

            field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),

        ),

        migrations.AddField(

            model_name='wcs',

            name='legacy_urls',

            field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),

        ),

        migrations.AddField(

            model_name='welco',

            name='legacy_urls',

            field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),

        ),

    ]

import time

from django.contrib.postgres.fields import JSONField

    legacy_urls = JSONField(null=True, default=list, blank=True)

        if self.legacy_urls:

            as_dict['legacy_urls'] = self.legacy_urls

    def change_base_url(self, base_url):

        service_dict = self.as_dict()

        timestamp = datetime.datetime.now()

        legacy_urls = {

            'base_url': service_dict['base_url'],

            'timestamp': str(time.mktime(timestamp.timetuple()) + timestamp.microsecond / 1e6),

        }

        for url_key in (

            'saml-sp-metadata-url',

            'saml-idp-metadata-url',

            'backoffice-menu-url',

            'provisionning-url',

        ):

            if url_key in service_dict:

                legacy_urls[url_key] = service_dict[url_key]

        if not self.legacy_urls:

            self.legacy_urls = []

        self.legacy_urls.insert(0, legacy_urls)

        self.base_url = base_url

    assert mocked_call_command.mock_calls == [

        call('create_tenant', 'combo.dev.publik.love', legacy_hostname=None)

    ]

def test_hobo_deploy_with_legacy_urls(monkeypatch, tenant_base, mocker, skeleton_dir, tmp_path):

    from django.core.management import call_command

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

    requests_get = mocker.patch('requests.get')

    meta1 = '''<?xml version="1.0"?>

<EntityDescriptor xmlns="urn:oasis:names:tc:SAML:2.0:metadata"

    xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion"

    xmlns:ds="http://www.w3.org/2000/09/xmldsig#"

    entityID="http://passerelle.example.net/saml/metadata">

  <SPSSODescriptor

    AuthnRequestsSigned="true" WantAssertionsSigned="true"

    protocolSupportEnumeration="urn:oasis:names:tc:SAML:2.0:protocol">

    <KeyDescriptor use="signing">

      <ds:KeyInfo xmlns:ds="http://www.w3.org/2000/09/xmldsig#">

        <KeyValue  xmlns="http://www.w3.org/2000/09/xmldsig#">

    <RSAKeyValue>

        <Modulus>nJpkBznHNbvE+RAC6mU+NPQnIWs8gFNCm6I3FPcUKYpaJbXaurJ4cJgvnaEiqIXPQDcbHxuLeCbYbId9yascWZirvQbh8d/r+Vv+24bPG++9gW+i3Nnz1VW8V+z0b+puHWvM/FjJjBNJgWkI38gaupz47U6/02CtWx00stitiwk=</Modulus>

        <Exponent>AQAB</Exponent>

    </RSAKeyValue>

</KeyValue>

      </ds:KeyInfo>

    </KeyDescriptor>

    <KeyDescriptor use="encryption">

      <ds:KeyInfo xmlns:ds="http://www.w3.org/2000/09/xmldsig#">

        <KeyValue  xmlns="http://www.w3.org/2000/09/xmldsig#">

    <RSAKeyValue>

        <Modulus>3BxSiAzGvY1Yuqa31L7Zr2WHM/8cn5oX+Q6A2SYgzjuvAgnWyizN8YgW/fHR4G7MtkmZ5RFJLXfcSLwbUfpFHV6KO1ikbgViYuFempM+SWtjqEI7ribm9GaI5kUzHJZBrH3/Q9XAd9/GLLALxurGjbKDeLfc0D+7el26g4sYmA8=</Modulus>

        <Exponent>AQAB</Exponent>

    </RSAKeyValue>

</KeyValue>

      </ds:KeyInfo>

    </KeyDescriptor>

    <SingleLogoutService

      Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"

      Location="http://passerelle.example.net/saml/singleLogout"

      ResponseLocation="http://passerelle.example.net/saml/singleLogoutReturn" />

    <SingleLogoutService

      Binding="urn:oasis:names:tc:SAML:2.0:bindings:SOAP"

      Location="http://passerelle.example.net/saml/singleLogoutSOAP" />

    <ManageNameIDService

      Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"

      Location="http://passerelle.example.net/saml/manageNameId"

      ResponseLocation="http://passerelle.example.net/saml/manageNameIdReturn" />

    <ManageNameIDService

      Binding="urn:oasis:names:tc:SAML:2.0:bindings:SOAP"

      Location="http://passerelle.example.net/saml/manageNameIdSOAP" />

    <AssertionConsumerService isDefault="true" index="0"

      Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Artifact"

      Location="http://passerelle.example.net/saml/assertionConsumerArtifact" />

    <AssertionConsumerService index="1"

      Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST"

      Location="http://passerelle.example.net/saml/assertionConsumerPost" />

    <AssertionConsumerService index="2"

      Binding="urn:oasis:names:tc:SAML:2.0:bindings:PAOS"

      Location="http://passerelle.example.net/saml/assertionConsumerSOAP" />

    <AssertionConsumerService index="3"

      Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"

      Location="http://passerelle.example.net/saml/assertionConsumerRedirect" />

  </SPSSODescriptor>

</EntityDescriptor>'''

    meta2 = meta1.replace('passerelle.example.net', 'new-passerelle.example.net')

    monkeypatch.setattr(HoboDeployCommand, 'backoff_factor', 0.0001)

    side_effect_iter = iter([meta1, meta2])

    def side_effect(*args, **kwargs):

        for v in side_effect_iter:

            m = mock.Mock()

            m.text = v

            return m

    requests_get.side_effect = side_effect

    def hobo_json(env_dict):

        with tempfile.NamedTemporaryFile(mode='w', dir=str(tmp_path), delete=False) as hobo_json:

            hobo_json_content = json.dumps(env_dict)

            hobo_json.write(hobo_json_content)

            return hobo_json.name

    env = {

        'services': [

            {

                'service-id': 'authentic',

                'slug': 'test',

                'title': 'Test',

                'this': True,

                'secret_key': '12345',

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

                'variables': {

                    'other_variable': 'bar',

                },

            },

            {

                'service-id': 'passerelle',

                'slug': 'passerelle',

                'title': u'Passerelle',

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

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

            },

        ],

        'users': [],

        'profile': {'fields': []},

    }

    with mock.patch('hobo.agent.authentic2.provisionning.notify_agents'):

        call_command('hobo_deploy', 'http://sso.example.net', hobo_json(env))

    from hobo.multitenant.middleware import TenantMiddleware

    tenants = list(TenantMiddleware.get_tenants())

    assert len(tenants) == 1

    tenant = tenants[0]

    assert tenant.domain_url == 'sso.example.net'

    assert tenant.schema_name == 'sso_example_net'

    tenant_directory = tenant.get_directory()

    assert tenant_directory == os.path.join(tenant_base, tenant.domain_url)

    assert os.path.exists(os.path.join(tenant_directory, 'saml.crt'))

    assert os.path.exists(os.path.join(tenant_directory, 'saml.key'))

    from tenant_schemas.utils import tenant_context

    with tenant_context(tenant):

        # SAML checks

        from authentic2.saml.models import LibertyProvider

        assert LibertyProvider.objects.count() == 1

        provider = LibertyProvider.objects.first()

        provider_id = provider.pk

        assert provider.entity_id == 'http://passerelle.example.net/saml/metadata'

        assert provider.metadata == meta1

    new_env = {

        'services': [

            {

                'service-id': 'authentic',

                'slug': 'test',

                'title': 'Test',

                'this': True,

                'secret_key': '12345',

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

                'variables': {

                    'other_variable': 'bar',

                },

            },

            {

                'service-id': 'passerelle',

                'slug': 'passerelle',

                'title': u'Passerelle',

                'base_url': 'http://new-passerelle.example.net',

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

                'legacy_urls': [

                    {

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

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

                    }

                ],

            },

        ],

        'users': [],

        'profile': {'fields': []},

    }

    with mock.patch('hobo.agent.authentic2.provisionning.notify_agents'):

        call_command('hobo_deploy', '--ignore-timestamp', 'http://sso.example.net', hobo_json(new_env))

    # check that liberty provider is updated

    with tenant_context(tenant):

        from authentic2.saml.models import LibertyProvider

        assert LibertyProvider.objects.count() == 1

        provider = LibertyProvider.objects.first()

        assert provider.metadata == meta2

        assert provider.entity_id == 'http://new-passerelle.example.net/saml/metadata'

        assert provider.pk == provider_id

    tenants = [

        make_tenant('tenant1.example.net'),

        make_tenant('hobo2.example.net'),

        make_tenant('hobo3.example.net'),

    ]

            if os.path.exists(t.get_directory()):

                t.delete(True)

    # URL change in interco portal

    with tenant_context(hobo1):

        combo = Combo.objects.get(slug='portal')

        assert combo.base_url == 'http://combo1.example.net/'

        combo.change_base_url('http://new-combo1.example.net')

        combo.save()

        # check the interco hobo json

        hobo_json = get_hobo_json()

        for service in hobo_json['services']:

            if service['slug'] == 'portal':

                assert service['base_url'] == 'http://new-combo1.example.net/'

                assert len(service['legacy_urls']) == 1

                assert service['legacy_urls'][0]['base_url'] == 'http://combo1.example.net/'

                break

        else:

            assert False, "no portal found"

    # inform coll2 about interco environment

    HoboDeployCommand().handle(hobo2.base_url, get_hobo_json_filename(hobo1))

    with tenant_context(hobo2):

        # no extra combo created in coll 2

        assert Combo.objects.filter().count() == 2

        # interco portal url changed

        combo = Combo.objects.get(slug='_interco_portal')

        assert combo.base_url == 'http://new-combo1.example.net/'

        assert len(combo.legacy_urls) == 1

        assert combo.legacy_urls[0]['base_url'] == 'http://combo1.example.net/'

    # inform coll2 about interco environment a second time, check that nothing changes

    HoboDeployCommand().handle(hobo2.base_url, get_hobo_json_filename(hobo1))

    with tenant_context(hobo2):

        assert Combo.objects.filter().count() == 2

        combo = Combo.objects.get(slug='_interco_portal')

        assert combo.base_url == 'http://new-combo1.example.net/'

        assert len(combo.legacy_urls) == 1

        assert combo.legacy_urls[0]['base_url'] == 'http://combo1.example.net/'

    # URL change in coll2 portal

    with tenant_context(hobo2):

        combo = Combo.objects.get(slug='portal')

        assert combo.base_url == 'http://combo2.example.net/'

        combo.change_base_url('http://new-combo2.example.net')

        combo.save()

        # check the coll2 hobo json

        hobo_json = get_hobo_json()

        for service in hobo_json['services']:

            if service['slug'] == 'portal':

                assert service['base_url'] == 'http://new-combo2.example.net/'

                assert len(service['legacy_urls']) == 1

                assert service['legacy_urls'][0]['base_url'] == 'http://combo2.example.net/'

                break

        else:

            assert False, "no portal found"

    # inform interco about coll2 environment

    HoboDeployCommand().handle(hobo1.base_url, get_hobo_json_filename(hobo2))

    with tenant_context(hobo1):

        # no extra combo created in interco

        assert Combo.objects.filter().count() == 3

        # coll2 portal url changed

        combo = Combo.objects.get(slug='_hobo-coll2_portal')

        assert combo.base_url == 'http://new-combo2.example.net/'

        assert len(combo.legacy_urls) == 1

        assert combo.legacy_urls[0]['base_url'] == 'http://combo2.example.net/'

    # URL change on the primary hobo

    with tenant_context(hobo1):

        hobo = Hobo.objects.get(slug='hobo')

        assert hobo.base_url == 'http://tenant1.example.net/'

        hobo.change_base_url('http://new-tenant1.example.net')

        hobo.save()

        # check the interco hobo json

        hobo_json = get_hobo_json()

        for service in hobo_json['services']:

            if service['slug'] == 'hobo':

                assert service['base_url'] == 'http://new-tenant1.example.net/'

                assert len(service['legacy_urls']) == 1

                assert service['legacy_urls'][0]['base_url'] == 'http://tenant1.example.net/'

                break

        else:

            assert False, 'no hobo found'

    # inform coll2 about interco environment

    HoboDeployCommand().handle(hobo2.base_url, get_hobo_json_filename(hobo1))

    with tenant_context(hobo2):

        # no extra hobo created in coll 2

        assert Hobo.objects.count() == 3

        # interco hobo url changed

        hobo = Hobo.objects.get(slug='_interco_hobo')

        assert hobo.base_url == 'http://new-tenant1.example.net/'

        assert len(hobo.legacy_urls) == 1

        assert hobo.legacy_urls[0]['base_url'] == 'http://tenant1.example.net/'

    # URL change on coll2 hobo (initiated by the interco tenant)

    with tenant_context(hobo1):

        hobo2 = Hobo.objects.get(slug='hobo-coll2')

        assert hobo2.base_url == 'http://hobo2.example.net/'

        hobo2.change_base_url('http://new-hobo2.example.net')

        hobo2.save()

    # inform coll2 about interco environment

    HoboDeployCommand().handle(hobo2.base_url, get_hobo_json_filename(hobo1))

    hobo2 = TenantMiddleware.get_tenant_by_hostname('new-hobo2.example.net')

    with tenant_context(hobo2):

        # no extra hobo created in coll 2

        assert Hobo.objects.count() == 3

        # coll2 hobo url changed

        hobo = Hobo.objects.get(slug='hobo')

        assert hobo.base_url == 'http://new-hobo2.example.net/'

        assert len(hobo.legacy_urls) == 1

        assert hobo.legacy_urls[0]['base_url'] == 'http://hobo2.example.net/'

{

    "services": [

	{

	    "service-id": "chrono",

	    "base_url": "https://new-chrono.dev.publik.love/",

	    "slug": "agendas",

	    "title": "CHRONO",

	    "secret_key": "123",

	    "template_name": "import_me",

            "legacy_urls": [

                {

                    "base_url": "https://chrono.dev.publik.love/"

                }

            ]

	},

	{

	    "service-id": "wcs",

	    "base_url": "https://wcs.dev.publik.love/",

	    "slug": "eservices",

	    "title": "WCS"

	},

	{

	    "service-id": "hobo",

	    "base_url": "https://hobo.dev.publik.love/",

	    "slug": "hobo",

	    "title": "HOBO",

	    "secret_key": "123"

	},

	{

	    "service-id": "combo",

	    "base_url": "https://combo.dev.publik.love/",

	    "slug": "portal",

	    "title": "COMBO",

	    "secret_key": "123",

	    "template_name": "import_me"

	},

	{

	    "service-id": "authentic",

	    "base_url": "https://authentic.dev.publik.love/",

	    "slug": "idp",

	    "title": "A2",

	    "secret_key": "123"

	}

    ]

}

import pytest

from hobo.multitenant.middleware import TenantMiddleware, TenantNotFound

    def my_call_command(command, parameter, **kwargs):

        call_command(command, parameter, **kwargs)

def test_deploy_with_legacy_urls(db):

    command = load_command_class('hobo.agent.common', 'hobo_deploy')

    domain = 'chrono.dev.publik.love'

    command.handle('https://%s/' % domain, 'tests_schemas/env.json')

    assert_deployed(domain)

    tenant_directory = TenantMiddleware.get_tenant_by_hostname(domain).get_directory()

    # change domain

    new_domain = 'new-chrono.dev.publik.love'

    command.handle('https://%s/' % new_domain, 'tests_schemas/legacy_urls_chrono_env.json')

    assert_deployed(new_domain)

    # check old tenant is gone

    with pytest.raises(TenantNotFound):

        TenantMiddleware.get_tenant_by_hostname(domain)

    assert not os.path.exists(tenant_directory)