674 lines
31 KiB
Python
674 lines
31 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import json
|
|
|
|
import pytest
|
|
import requests
|
|
import lasso
|
|
|
|
from mock import patch, call, ANY
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.core.management import call_command
|
|
|
|
from tenant_schemas.utils import tenant_context
|
|
|
|
from authentic2.saml.models import LibertyProvider
|
|
from authentic2.a2_rbac.models import Role, RoleAttribute
|
|
from authentic2.a2_rbac.utils import get_default_ou
|
|
from authentic2.models import Attribute, AttributeValue
|
|
from django_rbac.utils import get_ou_model
|
|
|
|
from hobo.agent.authentic2.provisionning import provisionning
|
|
from hobo import signature
|
|
|
|
User = get_user_model()
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def test_provision_role(transactional_db, tenant, caplog):
|
|
with patch('hobo.agent.authentic2.provisionning.notify_agents') as notify_agents:
|
|
with tenant_context(tenant):
|
|
LibertyProvider.objects.create(ou=get_default_ou(), name='provider',
|
|
entity_id='http://provider.com',
|
|
protocol_conformance=lasso.PROTOCOL_SAML_2_0)
|
|
with provisionning:
|
|
role = Role.objects.create(name='coin', ou=get_default_ou())
|
|
|
|
assert notify_agents.call_count == 1
|
|
arg = notify_agents.call_args
|
|
assert arg == call(ANY)
|
|
arg = arg[0][0]
|
|
assert isinstance(arg, dict)
|
|
assert set(arg.keys()) == set([
|
|
'audience', '@type', 'objects', 'full'])
|
|
assert arg['audience'] == ['http://provider.com']
|
|
assert arg['@type'] == 'provision'
|
|
assert arg['full'] is False
|
|
objects = arg['objects']
|
|
assert isinstance(objects, dict)
|
|
assert set(objects.keys()) == set(['data', '@type'])
|
|
assert objects['@type'] == 'role'
|
|
data = objects['data']
|
|
assert isinstance(data, list)
|
|
assert len(data) == 1
|
|
o = data[0]
|
|
assert set(o.keys()) == set(['details', 'emails_to_members',
|
|
'description', 'uuid', 'name',
|
|
'slug', 'emails'])
|
|
assert o['details'] == ''
|
|
assert o['emails_to_members'] is True
|
|
assert o['emails'] == []
|
|
|
|
notify_agents.reset_mock()
|
|
emails = ['john.doe@example.com', 'toto@entrouvert.com']
|
|
with provisionning:
|
|
RoleAttribute.objects.create(
|
|
role=role, name='emails', kind='json',
|
|
value=json.dumps(emails))
|
|
RoleAttribute.objects.create(
|
|
role=role, name='emails_to_members', kind='json',
|
|
value=json.dumps(True))
|
|
|
|
assert notify_agents.call_count == 1
|
|
arg = notify_agents.call_args
|
|
assert arg == call(ANY)
|
|
arg = arg[0][0]
|
|
assert isinstance(arg, dict)
|
|
assert set(arg.keys()) == set([
|
|
'audience', '@type', 'objects', 'full'])
|
|
assert arg['audience'] == ['http://provider.com']
|
|
assert arg['@type'] == 'provision'
|
|
assert arg['full'] is False
|
|
objects = arg['objects']
|
|
assert isinstance(objects, dict)
|
|
assert set(objects.keys()) == set(['data', '@type'])
|
|
assert objects['@type'] == 'role'
|
|
data = objects['data']
|
|
assert isinstance(data, list)
|
|
assert len(data) == 1
|
|
o = data[0]
|
|
assert set(o.keys()) == set(['details', 'emails_to_members',
|
|
'description', 'uuid', 'name',
|
|
'slug', 'emails'])
|
|
assert o['details'] == ''
|
|
assert o['emails_to_members'] is True
|
|
assert o['emails'] == emails
|
|
|
|
notify_agents.reset_mock()
|
|
with provisionning:
|
|
role.delete()
|
|
|
|
assert notify_agents.call_count == 1
|
|
arg = notify_agents.call_args
|
|
assert arg == call(ANY)
|
|
arg = arg[0][0]
|
|
assert isinstance(arg, dict)
|
|
assert set(arg.keys()) == set([
|
|
'audience', '@type', 'objects', 'full'])
|
|
assert arg['audience'] == ['http://provider.com']
|
|
assert arg['@type'] == 'deprovision'
|
|
assert arg['full'] is False
|
|
objects = arg['objects']
|
|
assert isinstance(objects, dict)
|
|
assert set(objects.keys()) == set(['data', '@type'])
|
|
assert objects['@type'] == 'role'
|
|
data = objects['data']
|
|
assert isinstance(data, list)
|
|
assert len(data) == 1
|
|
o = data[0]
|
|
assert set(o.keys()) == set(['uuid'])
|
|
|
|
|
|
def test_provision_user(transactional_db, tenant, caplog):
|
|
with patch('hobo.agent.authentic2.provisionning.notify_agents') as notify_agents:
|
|
with tenant_context(tenant):
|
|
service = LibertyProvider.objects.create(ou=get_default_ou(), name='provider',
|
|
entity_id='http://provider.com',
|
|
protocol_conformance=lasso.PROTOCOL_SAML_2_0)
|
|
role = Role.objects.create(name='coin', service=service, ou=get_default_ou())
|
|
role.attributes.create(kind='string', name='is_superuser', value='true')
|
|
role2 = Role.objects.create(name='zob', service=service, ou=get_default_ou())
|
|
role2.attributes.create(kind='json', name='emails', value='["zob@example.net"]')
|
|
child_role = Role.objects.create(name='child', ou=get_default_ou())
|
|
notify_agents.reset_mock()
|
|
attribute = Attribute.objects.create(label='Code postal', name='code_postal',
|
|
kind='string')
|
|
with provisionning:
|
|
user1 = User.objects.create(username=u'Étienne',
|
|
email='etienne.dugenou@example.net',
|
|
first_name=u'Étienne',
|
|
last_name=u'Dugenou',
|
|
ou=get_default_ou())
|
|
user2 = User.objects.create(username=u'john.doe',
|
|
email='iohn.doe@example.net',
|
|
first_name=u'John',
|
|
last_name=u'Doe',
|
|
is_active=False,
|
|
ou=get_default_ou())
|
|
role2.members.add(user2)
|
|
users = {user.uuid: user for user in [user1, user2]}
|
|
assert notify_agents.call_count == 2
|
|
assert notify_agents.call_args_list[0][0][0]['objects']['@type'] == 'role'
|
|
arg = notify_agents.call_args
|
|
assert arg == call(ANY)
|
|
arg = arg[0][0]
|
|
assert isinstance(arg, dict)
|
|
assert set(arg.keys()) == set([
|
|
'issuer', 'audience', '@type', 'objects', 'full'])
|
|
assert arg['issuer'] == \
|
|
'http://%s/idp/saml2/metadata' % tenant.domain_url
|
|
assert arg['audience'] == ['http://provider.com']
|
|
assert arg['@type'] == 'provision'
|
|
assert arg['full'] is False
|
|
objects = arg['objects']
|
|
assert isinstance(objects, dict)
|
|
assert set(objects.keys()) == set(['data', '@type'])
|
|
assert objects['@type'] == 'user'
|
|
data = objects['data']
|
|
assert isinstance(data, list)
|
|
assert len(data) == 2
|
|
assert len(set([o['uuid'] for o in data])) == 2
|
|
for o in data:
|
|
assert set(o.keys()) >= set(['uuid', 'username', 'first_name', 'is_superuser',
|
|
'last_name', 'email', 'roles'])
|
|
assert o['uuid'] in users
|
|
user = users[o['uuid']]
|
|
assert o['username'] == user.username
|
|
assert o['first_name'] == user.first_name
|
|
assert o['last_name'] == user.last_name
|
|
assert o['email'] == user.email
|
|
assert o['is_active'] is user.is_active
|
|
assert o['roles'] == [{'name': r.name, 'slug': r.slug, 'uuid': r.uuid} for r in
|
|
user.roles.all()]
|
|
assert o['is_superuser'] is False
|
|
|
|
notify_agents.reset_mock()
|
|
attribute.set_value(user1, '13400')
|
|
user1.is_superuser = True
|
|
with provisionning:
|
|
user1.save()
|
|
user2.save()
|
|
|
|
assert notify_agents.call_count == 1
|
|
arg = notify_agents.call_args
|
|
assert arg == call(ANY)
|
|
arg = arg[0][0]
|
|
assert isinstance(arg, dict)
|
|
assert set(arg.keys()) == set([
|
|
'issuer', 'audience', '@type', 'objects', 'full'])
|
|
assert arg['issuer'] == \
|
|
'http://%s/idp/saml2/metadata' % tenant.domain_url
|
|
assert arg['audience'] == ['http://provider.com']
|
|
assert arg['@type'] == 'provision'
|
|
assert arg['full'] is False
|
|
objects = arg['objects']
|
|
assert isinstance(objects, dict)
|
|
assert set(objects.keys()) == set(['data', '@type'])
|
|
assert objects['@type'] == 'user'
|
|
data = objects['data']
|
|
assert isinstance(data, list)
|
|
assert len(data) == 2
|
|
for o, user in zip(data, [user1, user2]):
|
|
assert set(o.keys()) >= set(['code_postal', 'uuid', 'username', 'first_name',
|
|
'is_superuser', 'last_name', 'email', 'roles'])
|
|
assert o['uuid'] == user.uuid
|
|
assert o['username'] == user.username
|
|
assert o['first_name'] == user.first_name
|
|
assert o['last_name'] == user.last_name
|
|
assert o['email'] == user.email
|
|
assert o['roles'] == [{'name': r.name, 'slug': r.slug, 'uuid': r.uuid} for r in
|
|
user.roles.all()]
|
|
assert o['code_postal'] is None or o['code_postal'] == '13400'
|
|
assert o['is_superuser'] is user.is_superuser
|
|
|
|
# test a service in a second OU also get the provisionning message
|
|
notify_agents.reset_mock()
|
|
ou2 = get_ou_model().objects.create(name=u'ou2', slug=u'ou2')
|
|
LibertyProvider.objects.create(ou=ou2, name='provider2',
|
|
slug='provider2',
|
|
entity_id='http://provider2.com',
|
|
protocol_conformance=lasso.PROTOCOL_SAML_2_0)
|
|
attribute.set_value(user1, '13500')
|
|
with provisionning:
|
|
user1.save()
|
|
user2.save()
|
|
|
|
assert notify_agents.call_count == 2
|
|
assert set(notify_agents.mock_calls[0][1][0]['audience'] +
|
|
notify_agents.mock_calls[1][1][0]['audience']) == set(['http://provider.com', 'http://provider2.com'])
|
|
ou2.delete()
|
|
|
|
notify_agents.reset_mock()
|
|
with provisionning:
|
|
AttributeValue.objects.get(attribute=attribute).delete()
|
|
|
|
assert notify_agents.call_count == 1
|
|
arg = notify_agents.call_args
|
|
assert arg == call(ANY)
|
|
arg = arg[0][0]
|
|
assert isinstance(arg, dict)
|
|
assert set(arg.keys()) == set([
|
|
'issuer', 'audience', '@type', 'objects', 'full'])
|
|
assert arg['issuer'] == \
|
|
'http://%s/idp/saml2/metadata' % tenant.domain_url
|
|
assert arg['audience'] == ['http://provider.com']
|
|
assert arg['@type'] == 'provision'
|
|
assert arg['full'] is False
|
|
objects = arg['objects']
|
|
assert isinstance(objects, dict)
|
|
assert set(objects.keys()) == set(['data', '@type'])
|
|
assert objects['@type'] == 'user'
|
|
data = objects['data']
|
|
assert isinstance(data, list)
|
|
assert len(data) == 1
|
|
for o in data:
|
|
assert set(o.keys()) >= set(['uuid', 'username', 'first_name',
|
|
'is_superuser', 'last_name', 'email', 'roles'])
|
|
assert o['uuid'] == user1.uuid
|
|
assert o['username'] == user1.username
|
|
assert o['first_name'] == user1.first_name
|
|
assert o['last_name'] == user1.last_name
|
|
assert o['email'] == user1.email
|
|
assert o['roles'] == []
|
|
assert o['is_superuser'] is True
|
|
|
|
user1.is_superuser = False
|
|
user1.save()
|
|
|
|
notify_agents.reset_mock()
|
|
with provisionning:
|
|
role.members.add(user1)
|
|
user2.save()
|
|
|
|
assert notify_agents.call_count == 2
|
|
assert notify_agents.call_args_list[0][0][0]['objects']['@type'] == 'role'
|
|
arg = notify_agents.call_args_list[1]
|
|
assert arg == call(ANY)
|
|
arg = arg[0][0]
|
|
assert isinstance(arg, dict)
|
|
assert set(arg.keys()) == set([
|
|
'issuer', 'audience', '@type', 'objects', 'full'])
|
|
assert arg['issuer'] == \
|
|
'http://%s/idp/saml2/metadata' % tenant.domain_url
|
|
assert arg['audience'] == ['http://provider.com']
|
|
assert arg['@type'] == 'provision'
|
|
assert arg['full'] is False
|
|
objects = arg['objects']
|
|
assert isinstance(objects, dict)
|
|
assert set(objects.keys()) == set(['data', '@type'])
|
|
assert objects['@type'] == 'user'
|
|
data = objects['data']
|
|
assert isinstance(data, list)
|
|
assert len(data) == 2
|
|
for o in data:
|
|
assert set(o.keys()) >= set(['uuid', 'username', 'first_name',
|
|
'is_superuser', 'last_name', 'email', 'roles'])
|
|
assert o['uuid'] in users
|
|
user = users[o['uuid']]
|
|
assert o['uuid'] == user.uuid
|
|
assert o['username'] == user.username
|
|
assert o['first_name'] == user.first_name
|
|
assert o['last_name'] == user.last_name
|
|
assert o['email'] == user.email
|
|
assert o['roles'] == [{'name': r.name, 'slug': r.slug, 'uuid': r.uuid} for r in
|
|
user.roles.all()]
|
|
assert o['is_superuser'] is (user == user1)
|
|
assert len(set(x['uuid'] for x in notify_agents.call_args_list[1][0][0]['objects']['data'])) == 2
|
|
|
|
notify_agents.reset_mock()
|
|
with provisionning:
|
|
user1.roles.remove(role)
|
|
|
|
assert notify_agents.call_count == 2
|
|
arg = notify_agents.call_args
|
|
assert arg == call(ANY)
|
|
arg = arg[0][0]
|
|
assert isinstance(arg, dict)
|
|
assert set(arg.keys()) == set([
|
|
'issuer', 'audience', '@type', 'objects', 'full'])
|
|
assert arg['issuer'] == \
|
|
'http://%s/idp/saml2/metadata' % tenant.domain_url
|
|
assert arg['audience'] == ['http://provider.com']
|
|
assert arg['@type'] == 'provision'
|
|
assert arg['full'] is False
|
|
objects = arg['objects']
|
|
assert isinstance(objects, dict)
|
|
assert set(objects.keys()) == set(['data', '@type'])
|
|
assert objects['@type'] == 'user'
|
|
data = objects['data']
|
|
assert isinstance(data, list)
|
|
assert len(data) == 1
|
|
for o in data:
|
|
assert set(o.keys()) >= set(['uuid', 'username', 'first_name',
|
|
'is_superuser', 'last_name', 'email', 'roles'])
|
|
assert o['uuid'] == user1.uuid
|
|
assert o['username'] == user1.username
|
|
assert o['first_name'] == user1.first_name
|
|
assert o['last_name'] == user1.last_name
|
|
assert o['email'] == user1.email
|
|
assert o['roles'] == []
|
|
assert o['is_superuser'] is False
|
|
|
|
notify_agents.reset_mock()
|
|
with provisionning:
|
|
user1.roles.add(child_role)
|
|
child_role.add_parent(role)
|
|
|
|
assert notify_agents.call_count == 2
|
|
arg = notify_agents.call_args
|
|
assert arg == call(ANY)
|
|
arg = arg[0][0]
|
|
assert isinstance(arg, dict)
|
|
assert set(arg.keys()) == set([
|
|
'issuer', 'audience', '@type', 'objects', 'full'])
|
|
assert arg['issuer'] == \
|
|
'http://%s/idp/saml2/metadata' % tenant.domain_url
|
|
assert arg['audience'] == ['http://provider.com']
|
|
assert arg['@type'] == 'provision'
|
|
assert arg['full'] is False
|
|
objects = arg['objects']
|
|
assert isinstance(objects, dict)
|
|
assert set(objects.keys()) == set(['data', '@type'])
|
|
assert objects['@type'] == 'user'
|
|
data = objects['data']
|
|
assert isinstance(data, list)
|
|
assert len(data) == 1
|
|
for o in data:
|
|
assert set(o.keys()) >= set(['uuid', 'username', 'first_name',
|
|
'is_superuser', 'last_name', 'email', 'roles'])
|
|
assert o['uuid'] == user1.uuid
|
|
assert o['username'] == user1.username
|
|
assert o['first_name'] == user1.first_name
|
|
assert o['last_name'] == user1.last_name
|
|
assert o['email'] == user1.email
|
|
assert len(o['roles']) == 2
|
|
for r in o['roles']:
|
|
r1 = {
|
|
'uuid': role.uuid,
|
|
'name': role.name,
|
|
'slug': role.slug
|
|
}
|
|
r2 = {
|
|
'uuid': child_role.uuid,
|
|
'name': child_role.name,
|
|
'slug': child_role.slug
|
|
}
|
|
assert r == r1 or r == r2
|
|
assert len(set(r['uuid'] for r in o['roles'])) == 2
|
|
assert o['is_superuser'] is True
|
|
|
|
notify_agents.reset_mock()
|
|
with provisionning:
|
|
child_role.remove_parent(role)
|
|
|
|
assert notify_agents.call_count == 1
|
|
arg = notify_agents.call_args
|
|
assert arg == call(ANY)
|
|
arg = arg[0][0]
|
|
assert isinstance(arg, dict)
|
|
assert set(arg.keys()) == set([
|
|
'issuer', 'audience', '@type', 'objects', 'full'])
|
|
assert arg['issuer'] == \
|
|
'http://%s/idp/saml2/metadata' % tenant.domain_url
|
|
assert arg['audience'] == ['http://provider.com']
|
|
assert arg['@type'] == 'provision'
|
|
assert arg['full'] is False
|
|
objects = arg['objects']
|
|
assert isinstance(objects, dict)
|
|
assert set(objects.keys()) == set(['data', '@type'])
|
|
assert objects['@type'] == 'user'
|
|
data = objects['data']
|
|
assert isinstance(data, list)
|
|
assert len(data) == 1
|
|
for o in data:
|
|
assert set(o.keys()) >= set(['uuid', 'username', 'first_name',
|
|
'is_superuser', 'last_name', 'email', 'roles'])
|
|
assert o['uuid'] == user1.uuid
|
|
assert o['username'] == user1.username
|
|
assert o['first_name'] == user1.first_name
|
|
assert o['last_name'] == user1.last_name
|
|
assert o['email'] == user1.email
|
|
assert o['roles'] == [{
|
|
'uuid': child_role.uuid,
|
|
'name': child_role.name,
|
|
'slug': child_role.slug
|
|
}]
|
|
assert o['is_superuser'] is False
|
|
|
|
notify_agents.reset_mock()
|
|
ou2 = get_ou_model().objects.create(name=u'ou2', slug=u'ou2')
|
|
LibertyProvider.objects.create(ou=get_default_ou(), name='provider2',
|
|
slug='provider2',
|
|
entity_id='http://provider2.com',
|
|
protocol_conformance=lasso.PROTOCOL_SAML_2_0)
|
|
with provisionning:
|
|
user1.delete()
|
|
user2.delete()
|
|
assert notify_agents.call_count == 1
|
|
arg = notify_agents.call_args
|
|
assert arg == call(ANY)
|
|
arg = arg[0][0]
|
|
assert isinstance(arg, dict)
|
|
assert set(arg.keys()) == set([
|
|
'issuer', 'audience', '@type', 'objects', 'full'])
|
|
assert arg['issuer'] == \
|
|
'http://%s/idp/saml2/metadata' % tenant.domain_url
|
|
assert set(arg['audience']) == set(['http://provider.com', 'http://provider2.com'])
|
|
assert arg['@type'] == 'deprovision'
|
|
assert arg['full'] is False
|
|
objects = arg['objects']
|
|
assert isinstance(objects, dict)
|
|
assert set(objects.keys()) == set(['data', '@type'])
|
|
assert objects['@type'] == 'user'
|
|
data = objects['data']
|
|
assert isinstance(data, list)
|
|
assert len(data) == 2
|
|
assert len(set([o['uuid'] for o in data])) == 2
|
|
for o in data:
|
|
assert set(o.keys()) == set(['uuid'])
|
|
assert o['uuid'] in users
|
|
|
|
|
|
def test_provision_createsuperuser(transactional_db, tenant, caplog):
|
|
with tenant_context(tenant):
|
|
# create a provider so notification messages have an audience.
|
|
LibertyProvider.objects.create(ou=None, name='provider',
|
|
entity_id='http://provider.com',
|
|
protocol_conformance=lasso.PROTOCOL_SAML_2_0)
|
|
with patch('hobo.agent.authentic2.provisionning.notify_agents') as notify_agents:
|
|
call_command('createsuperuser', domain=tenant.domain_url,
|
|
username='coin', email='coin@coin.org', interactive=False)
|
|
assert notify_agents.call_count == 1
|
|
|
|
|
|
def test_command_hobo_provision(notify_agents, transactional_db, tenant, caplog):
|
|
with tenant_context(tenant):
|
|
ou = get_default_ou()
|
|
LibertyProvider.objects.create(ou=ou, name='provider',
|
|
entity_id='http://provider.com',
|
|
protocol_conformance=lasso.PROTOCOL_SAML_2_0)
|
|
for i in range(10):
|
|
Role.objects.create(name='role-%s' % i, ou=ou)
|
|
for i in range(10):
|
|
User.objects.create(username='user-%s' % i, first_name='John',
|
|
last_name='Doe %s' % i, ou=ou,
|
|
email='jone.doe-%s@example.com')
|
|
|
|
with tenant_context(tenant):
|
|
# call_command('tenant_command', 'hobo_provision', ...) doesn't work
|
|
# https://github.com/bernardopires/django-tenant-schemas/issues/495
|
|
# so we call the command from the tenant context.
|
|
call_command('hobo_provision', roles=True, users=True)
|
|
|
|
msg_1 = notify_agents.call_args_list[0][0][0]
|
|
msg_2 = notify_agents.call_args_list[1][0][0]
|
|
assert msg_1['@type'] == 'provision'
|
|
assert msg_1['full'] is True
|
|
assert msg_1['objects']['@type'] == 'role'
|
|
assert len(msg_1['objects']['data']) == 10
|
|
assert msg_2['@type'] == 'provision'
|
|
assert msg_2['full'] is False
|
|
assert msg_2['objects']['@type'] == 'user'
|
|
assert len(msg_2['objects']['data']) == 10
|
|
|
|
|
|
def test_middleware(notify_agents, app_factory, tenant, settings):
|
|
settings.HOBO_PROVISIONNING_SYNCHRONOUS = True
|
|
|
|
with tenant_context(tenant):
|
|
user = User.objects.create(username='john', ou=get_default_ou())
|
|
user.set_password('password')
|
|
user.save()
|
|
LibertyProvider.objects.create(ou=get_default_ou(),
|
|
name='provider',
|
|
entity_id='http://provider.com',
|
|
protocol_conformance=lasso.PROTOCOL_SAML_2_0)
|
|
assert notify_agents.call_count == 0
|
|
|
|
app = app_factory(tenant)
|
|
resp = app.get('/login/')
|
|
form = resp.form
|
|
form.set('username', 'john')
|
|
form.set('password', 'password')
|
|
resp = form.submit(name='login-password-submit').follow()
|
|
resp = resp.click('Your account')
|
|
resp = resp.click('Edit')
|
|
resp.form.set('first_name', 'John')
|
|
resp.form.set('last_name', 'Doe')
|
|
assert notify_agents.call_count == 0
|
|
resp = resp.form.submit().follow()
|
|
assert notify_agents.call_count == 1
|
|
|
|
|
|
def test_provision_using_http(transactional_db, tenant, settings, caplog):
|
|
with tenant_context(tenant):
|
|
# create providers so notification messages have an audience.
|
|
LibertyProvider.objects.create(ou=None, name='provider', slug='provider',
|
|
entity_id='http://example.org',
|
|
protocol_conformance=lasso.PROTOCOL_SAML_2_0)
|
|
LibertyProvider.objects.create(ou=None, name='provider2', slug='provider2',
|
|
entity_id='http://example.com',
|
|
protocol_conformance=lasso.PROTOCOL_SAML_2_0)
|
|
with patch('hobo.agent.authentic2.provisionning.notify_agents') as notify_agents:
|
|
call_command('createsuperuser', domain=tenant.domain_url,
|
|
username='coin', email='coin@coin.org', interactive=False)
|
|
assert notify_agents.call_count == 1
|
|
assert set(notify_agents.call_args[0][0]['audience']) == {'http://example.org', 'http://example.com'}
|
|
|
|
settings.HOBO_HTTP_PROVISIONNING = True
|
|
with patch('hobo.agent.authentic2.provisionning.notify_agents') as notify_agents:
|
|
call_command('createsuperuser', domain=tenant.domain_url,
|
|
username='coin2', email='coin2@coin.org', interactive=False)
|
|
assert notify_agents.call_count == 1
|
|
assert set(notify_agents.call_args[0][0]['audience']) == {'http://example.org', 'http://example.com'}
|
|
|
|
settings.HOBO_HTTP_PROVISIONNING = True
|
|
settings.KNOWN_SERVICES = {
|
|
'foo': {
|
|
'bar': {
|
|
'saml-sp-metadata-url': 'http://example.org',
|
|
'provisionning-url': 'http://example.org/__provision__/',
|
|
'orig': 'example.org',
|
|
'secret': 'xxx',
|
|
}
|
|
}
|
|
}
|
|
with patch('hobo.agent.authentic2.provisionning.notify_agents') as notify_agents:
|
|
with patch('hobo.agent.authentic2.provisionning.requests.put') as requests_put:
|
|
call_command('createsuperuser', domain=tenant.domain_url,
|
|
username='coin2', email='coin2@coin.org', interactive=False)
|
|
assert notify_agents.call_count == 1
|
|
assert notify_agents.call_args[0][0]['audience'] == ['http://example.com']
|
|
assert requests_put.call_count == 1
|
|
# cannot check audience passed to requests.put as it's the same
|
|
# dictionary that is altered afterwards and would thus also contain
|
|
# http://example.com.
|
|
|
|
settings.KNOWN_SERVICES['foo']['bar2'] = {
|
|
'saml-sp-metadata-url': 'http://example.com',
|
|
'provisionning-url': 'http://example.com/__provision__/',
|
|
'orig': 'example.com',
|
|
'secret': 'xxx',
|
|
}
|
|
with patch('hobo.agent.authentic2.provisionning.notify_agents') as notify_agents:
|
|
with patch('hobo.agent.authentic2.provisionning.requests.put') as requests_put:
|
|
call_command('createsuperuser', domain=tenant.domain_url,
|
|
username='coin2', email='coin2@coin.org', interactive=False)
|
|
assert notify_agents.call_count == 0
|
|
assert requests_put.call_count == 2
|
|
|
|
|
|
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
|
|
assert not resp.json['leftover_audience']
|
|
assert set(resp.json['reached_audience']) == {
|
|
'http://other.example.net/metadata/',
|
|
'http://more.example.net/metadata/'}
|
|
|
|
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
|
|
assert not resp.json['leftover_audience']
|
|
assert set(resp.json['reached_audience']) == {'http://other.example.net/metadata/'}
|
|
|
|
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
|
|
assert resp.json['err'] == 0
|
|
assert not resp.json['leftover_audience']
|
|
|
|
with patch('hobo.agent.authentic2.provisionning.requests.put') as requests_put:
|
|
requests_put.side_effect = requests.RequestException
|
|
resp = app.post_json(signature.sign_url('/api/provision/?orig=%s' % orig,
|
|
key), {'role_uuid': role.uuid})
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['leftover_audience']
|