<NME> test_v3_federation.py <BEF> # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import copy import os import random import re import subprocess from testtools import matchers from unittest import mock import uuid import fixtures import flask import http.client from lxml import etree from oslo_serialization import jsonutils from oslo_utils import importutils import saml2 from saml2 import saml from saml2 import sigver import urllib xmldsig = importutils.try_import("saml2.xmldsig") if not xmldsig: xmldsig = importutils.try_import("xmldsig") from keystone.api._shared import authentication from keystone.api import auth as auth_api from keystone.common import driver_hints from keystone.common import provider_api from keystone.common import render_token import keystone.conf from keystone import exception from keystone.federation import idp as keystone_idp from keystone.models import token_model from keystone import notifications from keystone.tests import unit from keystone.tests.unit import core from keystone.tests.unit import federation_fixtures from keystone.tests.unit import ksfixtures from keystone.tests.unit import mapping_fixtures from keystone.tests.unit import test_v3 CONF = keystone.conf.CONF PROVIDERS = provider_api.ProviderAPIs ROOTDIR = os.path.dirname(os.path.abspath(__file__)) XMLDIR = os.path.join(ROOTDIR, 'saml2/') def dummy_validator(*args, **kwargs): pass class FederatedSetupMixin(object): ACTION = 'authenticate' IDP = 'ORG_IDP' PROTOCOL = 'saml2' AUTH_METHOD = 'saml2' USER = 'user@ORGANIZATION' ASSERTION_PREFIX = 'PREFIX_' IDP_WITH_REMOTE = 'ORG_IDP_REMOTE' REMOTE_IDS = ['entityID_IDP1', 'entityID_IDP2'] REMOTE_ID_ATTR = uuid.uuid4().hex UNSCOPED_V3_SAML2_REQ = { "identity": { "methods": [AUTH_METHOD], AUTH_METHOD: { "identity_provider": IDP, "protocol": PROTOCOL } } } def _check_domains_are_valid(self, token): domain = PROVIDERS.resource_api.get_domain(self.idp['domain_id']) self.assertEqual(domain['id'], token['user']['domain']['id']) self.assertEqual(domain['name'], token['user']['domain']['name']) def _project(self, project): return (project['id'], project['name']) def _roles(self, roles): return set([(r['id'], r['name']) for r in roles]) def _check_projects_and_roles(self, token, roles, projects): """Check whether the projects and the roles match.""" token_roles = token.get('roles') if token_roles is None: raise AssertionError('Roles not found in the token') token_roles = self._roles(token_roles) roles_ref = self._roles(roles) self.assertEqual(token_roles, roles_ref) token_projects = token.get('project') if token_projects is None: raise AssertionError('Projects not found in the token') token_projects = self._project(token_projects) projects_ref = self._project(projects) self.assertEqual(token_projects, projects_ref) def _check_scoped_token_attributes(self, token): for obj in ('user', 'catalog', 'expires_at', 'issued_at', 'methods', 'roles'): self.assertIn(obj, token) os_federation = token['user']['OS-FEDERATION'] self.assertIn('groups', os_federation) self.assertIn('identity_provider', os_federation) self.assertIn('protocol', os_federation) self.assertThat(os_federation, matchers.HasLength(3)) self.assertEqual(self.IDP, os_federation['identity_provider']['id']) self.assertEqual(self.PROTOCOL, os_federation['protocol']['id']) def _check_project_scoped_token_attributes(self, token, project_id): self.assertEqual(project_id, token['project']['id']) self._check_scoped_token_attributes(token) def _check_domain_scoped_token_attributes(self, token, domain_id): self.assertEqual(domain_id, token['domain']['id']) self._check_scoped_token_attributes(token) def assertValidMappedUser(self, token): """Check if user object meets all the criteria.""" user = token['user'] self.assertIn('id', user) self.assertIn('name', user) self.assertIn('domain', user) self.assertIn('groups', user['OS-FEDERATION']) self.assertIn('identity_provider', user['OS-FEDERATION']) self.assertIn('protocol', user['OS-FEDERATION']) # Make sure user_name is url safe self.assertEqual(urllib.parse.quote(user['name']), user['name']) def _issue_unscoped_token(self, idp=None, assertion='EMPLOYEE_ASSERTION', environment=None): environment = environment or {} environment.update(getattr(mapping_fixtures, assertion)) with self.make_request(environ=environment): if idp is None: idp = self.IDP r = authentication.federated_authenticate_for_token( protocol_id=self.PROTOCOL, identity_provider=idp) return r def idp_ref(self, id=None): idp = { 'id': id or uuid.uuid4().hex, 'enabled': True, 'description': uuid.uuid4().hex } return idp def proto_ref(self, mapping_id=None): proto = { 'id': uuid.uuid4().hex, 'mapping_id': mapping_id or uuid.uuid4().hex } return proto def mapping_ref(self, rules=None): return { 'id': uuid.uuid4().hex, 'rules': rules or self.rules['rules'] } def _scope_request(self, unscoped_token_id, scope, scope_id): return { 'auth': { 'identity': { 'methods': [ 'token' ], 'token': { 'id': unscoped_token_id } }, 'scope': { scope: { 'id': scope_id } } } } def _inject_assertion(self, variant): assertion = getattr(mapping_fixtures, variant) flask.request.environ.update(assertion) def load_federation_sample_data(self): """Inject additional data.""" # Create and add domains self.domainA = unit.new_domain_ref() PROVIDERS.resource_api.create_domain( self.domainA['id'], self.domainA ) self.domainB = unit.new_domain_ref() PROVIDERS.resource_api.create_domain( self.domainB['id'], self.domainB ) self.domainC = unit.new_domain_ref() PROVIDERS.resource_api.create_domain( self.domainC['id'], self.domainC ) self.domainD = unit.new_domain_ref() PROVIDERS.resource_api.create_domain( self.domainD['id'], self.domainD ) # Create and add projects self.proj_employees = unit.new_project_ref( domain_id=self.domainA['id']) PROVIDERS.resource_api.create_project( self.proj_employees['id'], self.proj_employees ) self.proj_customers = unit.new_project_ref( domain_id=self.domainA['id']) PROVIDERS.resource_api.create_project( self.proj_customers['id'], self.proj_customers ) self.project_all = unit.new_project_ref( domain_id=self.domainA['id']) PROVIDERS.resource_api.create_project( self.project_all['id'], self.project_all ) self.project_inherited = unit.new_project_ref( domain_id=self.domainD['id']) PROVIDERS.resource_api.create_project( self.project_inherited['id'], self.project_inherited ) # Create and add groups self.group_employees = unit.new_group_ref(domain_id=self.domainA['id']) self.group_employees = ( PROVIDERS.identity_api.create_group(self.group_employees)) self.group_customers = unit.new_group_ref(domain_id=self.domainA['id']) self.group_customers = ( PROVIDERS.identity_api.create_group(self.group_customers)) self.group_admins = unit.new_group_ref(domain_id=self.domainA['id']) self.group_admins = PROVIDERS.identity_api.create_group( self.group_admins ) # Create and add roles self.role_employee = unit.new_role_ref() PROVIDERS.role_api.create_role( self.role_employee['id'], self.role_employee ) self.role_customer = unit.new_role_ref() PROVIDERS.role_api.create_role( self.role_customer['id'], self.role_customer ) self.role_admin = unit.new_role_ref() PROVIDERS.role_api.create_role(self.role_admin['id'], self.role_admin) # Employees can access # * proj_employees # * project_all PROVIDERS.assignment_api.create_grant( self.role_employee['id'], group_id=self.group_employees['id'], project_id=self.proj_employees['id'] ) PROVIDERS.assignment_api.create_grant( self.role_employee['id'], group_id=self.group_employees['id'], project_id=self.project_all['id'] ) # Customers can access # * proj_customers PROVIDERS.assignment_api.create_grant( self.role_customer['id'], group_id=self.group_customers['id'], project_id=self.proj_customers['id'] ) # Admins can access: # * proj_customers # * proj_employees # * project_all PROVIDERS.assignment_api.create_grant( self.role_admin['id'], group_id=self.group_admins['id'], project_id=self.proj_customers['id'] ) PROVIDERS.assignment_api.create_grant( self.role_admin['id'], group_id=self.group_admins['id'], project_id=self.proj_employees['id'] ) PROVIDERS.assignment_api.create_grant( self.role_admin['id'], group_id=self.group_admins['id'], project_id=self.project_all['id'] ) # Customers can access: # * domain A PROVIDERS.assignment_api.create_grant( self.role_customer['id'], group_id=self.group_customers['id'], domain_id=self.domainA['id'] ) # Customers can access projects via inheritance: # * domain D PROVIDERS.assignment_api.create_grant( self.role_customer['id'], group_id=self.group_customers['id'], domain_id=self.domainD['id'], inherited_to_projects=True ) # Employees can access: # * domain A # * domain B PROVIDERS.assignment_api.create_grant( self.role_employee['id'], group_id=self.group_employees['id'], domain_id=self.domainA['id'] ) PROVIDERS.assignment_api.create_grant( self.role_employee['id'], group_id=self.group_employees['id'], domain_id=self.domainB['id'] ) # Admins can access: # * domain A # * domain B # * domain C PROVIDERS.assignment_api.create_grant( self.role_admin['id'], group_id=self.group_admins['id'], domain_id=self.domainA['id'] ) PROVIDERS.assignment_api.create_grant( self.role_admin['id'], group_id=self.group_admins['id'], domain_id=self.domainB['id'] ) PROVIDERS.assignment_api.create_grant( self.role_admin['id'], group_id=self.group_admins['id'], domain_id=self.domainC['id'] ) self.rules = { 'rules': [ { 'local': [ { 'group': { 'id': self.group_employees['id'] } }, { 'user': { 'name': '{0}', 'id': '{1}' } } ], 'remote': [ { 'type': 'UserName' }, { 'type': 'Email', }, { 'type': 'orgPersonType', 'any_one_of': [ 'Employee' ] } ] }, { 'local': [ { 'group': { 'id': self.group_employees['id'] } }, { 'user': { 'name': '{0}', 'id': '{1}' } } ], 'remote': [ { 'type': self.ASSERTION_PREFIX + 'UserName' }, { 'type': self.ASSERTION_PREFIX + 'Email', }, { 'type': self.ASSERTION_PREFIX + 'orgPersonType', 'any_one_of': [ 'SuperEmployee' ] } ] }, { 'local': [ { 'group': { 'id': self.group_customers['id'] } }, { 'user': { 'name': '{0}', 'id': '{1}' } } ], 'remote': [ { 'type': 'UserName' }, { 'type': 'Email' }, { 'type': 'orgPersonType', 'any_one_of': [ 'Customer' ] } ] }, { 'local': [ { 'group': { 'id': self.group_admins['id'] } }, { 'group': { 'id': self.group_employees['id'] } }, { 'group': { 'id': self.group_customers['id'] } }, { 'user': { 'name': '{0}', 'id': '{1}' } } ], 'remote': [ { 'type': 'UserName' }, { 'type': 'Email' }, { 'type': 'orgPersonType', 'any_one_of': [ 'Admin', 'Chief' ] } ] }, { 'local': [ { 'group': { 'id': uuid.uuid4().hex } }, { 'group': { 'id': self.group_customers['id'] } }, { 'user': { 'name': '{0}', 'id': '{1}' } } ], 'remote': [ { 'type': 'UserName', }, { 'type': 'Email', }, { 'type': 'FirstName', 'any_one_of': [ 'Jill' ] }, { 'type': 'LastName', 'any_one_of': [ 'Smith' ] } ] }, { 'local': [ { 'group': { 'id': 'this_group_no_longer_exists' } }, { 'user': { 'name': '{0}', 'id': '{1}' } } ], 'remote': [ { 'type': 'UserName', }, { 'type': 'Email', }, { 'type': 'Email', 'any_one_of': [ 'testacct@example.com' ] }, { 'type': 'orgPersonType', 'any_one_of': [ 'Tester' ] } ] }, # rules with local group names { "local": [ { 'user': { 'name': '{0}', 'id': '{1}' } }, { "group": { "name": self.group_customers['name'], "domain": { "name": self.domainA['name'] } } } ], "remote": [ { 'type': 'UserName', }, { 'type': 'Email', }, { "type": "orgPersonType", "any_one_of": [ "CEO", "CTO" ], } ] }, { "local": [ { 'user': { 'name': '{0}', 'id': '{1}' } }, { "group": { "name": self.group_admins['name'], "domain": { "id": self.domainA['id'] } } } ], "remote": [ { "type": "UserName", }, { "type": "Email", }, { "type": "orgPersonType", "any_one_of": [ "Managers" ] } ] }, { "local": [ { "user": { "name": "{0}", "id": "{1}" } }, { "group": { "name": "NON_EXISTING", "domain": { "id": self.domainA['id'] } } } ], "remote": [ { "type": "UserName", }, { "type": "Email", }, { "type": "UserName", "any_one_of": [ "IamTester" ] } ] }, { "local": [ { "user": { "type": "local", "name": self.user['name'], "domain": { "id": self.user['domain_id'] } } }, { "group": { "id": self.group_customers['id'] } } ], "remote": [ { "type": "UserType", "any_one_of": [ "random" ] } ] }, { "local": [ { "user": { "type": "local", "name": self.user['name'], "domain": { "id": uuid.uuid4().hex } } } ], "remote": [ { "type": "Position", "any_one_of": [ "DirectorGeneral" ] } ] }, # rules for users with no groups { "local": [ { 'user': { 'name': '{0}', 'id': '{1}' } } ], "remote": [ { 'type': 'UserName', }, { 'type': 'Email', }, { 'type': 'orgPersonType', 'any_one_of': [ 'NoGroupsOrg' ] } ] } ] } # Add unused IdP first so it is indexed first (#1838592) self.dummy_idp = self.idp_ref() PROVIDERS.federation_api.create_idp( self.dummy_idp['id'], self.dummy_idp ) # Add IDP self.idp = self.idp_ref(id=self.IDP) PROVIDERS.federation_api.create_idp( self.idp['id'], self.idp ) # Add IDP with remote self.idp_with_remote = self.idp_ref(id=self.IDP_WITH_REMOTE) self.idp_with_remote['remote_ids'] = self.REMOTE_IDS PROVIDERS.federation_api.create_idp( self.idp_with_remote['id'], self.idp_with_remote ) # Add a mapping self.mapping = self.mapping_ref() PROVIDERS.federation_api.create_mapping( self.mapping['id'], self.mapping ) # Add protocols self.proto_saml = self.proto_ref(mapping_id=self.mapping['id']) self.proto_saml['id'] = self.PROTOCOL PROVIDERS.federation_api.create_protocol( self.idp['id'], self.proto_saml['id'], self.proto_saml ) # Add protocols IDP with remote PROVIDERS.federation_api.create_protocol( self.idp_with_remote['id'], self.proto_saml['id'], self.proto_saml ) # Add unused protocol to go with unused IdP (#1838592) self.proto_dummy = self.proto_ref(mapping_id=self.mapping['id']) PROVIDERS.federation_api.create_protocol( self.dummy_idp['id'], self.proto_dummy['id'], self.proto_dummy ) with self.make_request(): self.tokens = {} VARIANTS = ('EMPLOYEE_ASSERTION', 'CUSTOMER_ASSERTION', 'ADMIN_ASSERTION') for variant in VARIANTS: self._inject_assertion(variant) r = authentication.authenticate_for_token( self.UNSCOPED_V3_SAML2_REQ) self.tokens[variant] = r.id self.TOKEN_SCOPE_PROJECT_FROM_NONEXISTENT_TOKEN = ( self._scope_request( uuid.uuid4().hex, 'project', self.proj_customers['id'])) self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_EMPLOYEE = ( self._scope_request( self.tokens['EMPLOYEE_ASSERTION'], 'project', self.proj_employees['id'])) self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN = self._scope_request( self.tokens['ADMIN_ASSERTION'], 'project', self.proj_employees['id']) self.TOKEN_SCOPE_PROJECT_CUSTOMER_FROM_ADMIN = self._scope_request( self.tokens['ADMIN_ASSERTION'], 'project', self.proj_customers['id']) self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_CUSTOMER = ( self._scope_request( self.tokens['CUSTOMER_ASSERTION'], 'project', self.proj_employees['id'])) self.TOKEN_SCOPE_PROJECT_INHERITED_FROM_CUSTOMER = ( self._scope_request( self.tokens['CUSTOMER_ASSERTION'], 'project', self.project_inherited['id'])) self.TOKEN_SCOPE_DOMAIN_A_FROM_CUSTOMER = self._scope_request( self.tokens['CUSTOMER_ASSERTION'], 'domain', self.domainA['id']) self.TOKEN_SCOPE_DOMAIN_B_FROM_CUSTOMER = self._scope_request( self.tokens['CUSTOMER_ASSERTION'], 'domain', self.domainB['id']) self.TOKEN_SCOPE_DOMAIN_D_FROM_CUSTOMER = self._scope_request( self.tokens['CUSTOMER_ASSERTION'], 'domain', self.domainD['id']) self.TOKEN_SCOPE_DOMAIN_A_FROM_ADMIN = self._scope_request( self.tokens['ADMIN_ASSERTION'], 'domain', self.domainA['id']) self.TOKEN_SCOPE_DOMAIN_B_FROM_ADMIN = self._scope_request( self.tokens['ADMIN_ASSERTION'], 'domain', self.domainB['id']) self.TOKEN_SCOPE_DOMAIN_C_FROM_ADMIN = self._scope_request( self.tokens['ADMIN_ASSERTION'], 'domain', self.domainC['id']) class FederatedIdentityProviderTests(test_v3.RestfulTestCase): """A test class for Identity Providers.""" idp_keys = ['description', 'enabled'] default_body = {'description': None, 'enabled': True} def base_url(self, suffix=None): if suffix is not None: return '/OS-FEDERATION/identity_providers/' + str(suffix) return '/OS-FEDERATION/identity_providers' def _fetch_attribute_from_response(self, resp, parameter, assert_is_not_none=True): """Fetch single attribute from TestResponse object.""" result = resp.result.get(parameter) if assert_is_not_none: self.assertIsNotNone(result) return result def _create_and_decapsulate_response(self, body=None): """Create IdP and fetch it's random id along with entity.""" default_resp = self._create_default_idp(body=body) idp = self._fetch_attribute_from_response(default_resp, 'identity_provider') self.assertIsNotNone(idp) idp_id = idp.get('id') return (idp_id, idp) def _get_idp(self, idp_id): """Fetch IdP entity based on its id.""" url = self.base_url(suffix=idp_id) resp = self.get(url) return resp def _create_default_idp(self, body=None, expected_status=http.client.CREATED): """Create default IdP.""" url = self.base_url(suffix=uuid.uuid4().hex) if body is None: body = self._http_idp_input() resp = self.put(url, body={'identity_provider': body}, expected_status=expected_status) return resp def _http_idp_input(self): """Create default input dictionary for IdP data.""" body = self.default_body.copy() body['description'] = uuid.uuid4().hex return body def _assign_protocol_to_idp(self, idp_id=None, proto=None, url=None, mapping_id=None, validate=True, **kwargs): if url is None: url = self.base_url(suffix='%(idp_id)s/protocols/%(protocol_id)s') if idp_id is None: idp_id, _ = self._create_and_decapsulate_response() if proto is None: proto = uuid.uuid4().hex if mapping_id is None: mapping_id = uuid.uuid4().hex self._create_mapping(mapping_id) body = {'mapping_id': mapping_id} url = url % {'idp_id': idp_id, 'protocol_id': proto} resp = self.put(url, body={'protocol': body}, **kwargs) if validate: self.assertValidResponse(resp, 'protocol', dummy_validator, keys_to_check=['id', 'mapping_id'], ref={'id': proto, 'mapping_id': mapping_id}) return (resp, idp_id, proto) def _get_protocol(self, idp_id, protocol_id): url = '%s/protocols/%s' % (idp_id, protocol_id) url = self.base_url(suffix=url) r = self.get(url) return r def _create_mapping(self, mapping_id): mapping = mapping_fixtures.MAPPING_EPHEMERAL_USER mapping['id'] = mapping_id url = '/OS-FEDERATION/mappings/%s' % mapping_id self.put(url, body={'mapping': mapping}, expected_status=http.client.CREATED) def assertIdpDomainCreated(self, idp_id, domain_id): domain = PROVIDERS.resource_api.get_domain(domain_id) self.assertEqual(domain_id, domain['name']) self.assertIn(idp_id, domain['description']) def test_create_idp_without_domain_id(self): """Create the IdentityProvider entity associated to remote_ids.""" keys_to_check = list(self.idp_keys) body = self.default_body.copy() body['description'] = uuid.uuid4().hex resp = self._create_default_idp(body=body) self.assertValidResponse(resp, 'identity_provider', dummy_validator, keys_to_check=keys_to_check, ref=body) attr = self._fetch_attribute_from_response(resp, 'identity_provider') self.assertIdpDomainCreated(attr['id'], attr['domain_id']) def test_create_idp_with_domain_id(self): keys_to_check = list(self.idp_keys) keys_to_check.append('domain_id') body = self.default_body.copy() body['description'] = uuid.uuid4().hex domain = unit.new_domain_ref() PROVIDERS.resource_api.create_domain(domain['id'], domain) body['domain_id'] = domain['id'] resp = self._create_default_idp(body=body) self.assertValidResponse(resp, 'identity_provider', dummy_validator, keys_to_check=keys_to_check, ref=body) def test_create_idp_domain_id_none(self): keys_to_check = list(self.idp_keys) body = self.default_body.copy() body['description'] = uuid.uuid4().hex body['domain_id'] = None resp = self._create_default_idp(body=body) self.assertValidResponse(resp, 'identity_provider', dummy_validator, keys_to_check=keys_to_check, ref=body) attr = self._fetch_attribute_from_response(resp, 'identity_provider') self.assertIdpDomainCreated(attr['id'], attr['domain_id']) def test_conflicting_idp_cleans_up_auto_generated_domain(self): # NOTE(lbragstad): Create an identity provider, save its ID, and count # the number of domains. resp = self._create_default_idp() idp_id = resp.json_body['identity_provider']['id'] domains = PROVIDERS.resource_api.list_domains() number_of_domains = len(domains) # Create an identity provider with the same ID to intentionally cause a # conflict, this is going to result in a domain getting created for the # new identity provider. The domain for the new identity provider is # going to be created before the conflict is raised from the database # layer. This makes sure the domain is cleaned up after a Conflict is # detected. resp = self.put( self.base_url(suffix=idp_id), body={'identity_provider': self.default_body.copy()}, expected_status=http.client.CONFLICT ) domains = PROVIDERS.resource_api.list_domains() self.assertEqual(number_of_domains, len(domains)) def test_conflicting_idp_does_not_delete_existing_domain(self): # Create a new domain domain = unit.new_domain_ref() PROVIDERS.resource_api.create_domain(domain['id'], domain) # Create an identity provider and specify the domain body = self.default_body.copy() body['description'] = uuid.uuid4().hex body['domain_id'] = domain['id'] resp = self._create_default_idp(body=body) idp = resp.json_body['identity_provider'] idp_id = idp['id'] self.assertEqual(idp['domain_id'], domain['id']) # Create an identity provider with the same domain and ID to ensure a # Conflict is raised and then to verify the existing domain not deleted # below body = self.default_body.copy() body['domain_id'] = domain['id'] resp = self.put( self.base_url(suffix=idp_id), body={'identity_provider': body}, expected_status=http.client.CONFLICT ) # Make sure the domain specified in the second request was not deleted, # since it wasn't auto-generated self.assertIsNotNone(PROVIDERS.resource_api.get_domain(domain['id'])) def test_create_multi_idp_to_one_domain(self): # create domain and add domain_id to keys to check domain = unit.new_domain_ref() PROVIDERS.resource_api.create_domain(domain['id'], domain) keys_to_check = list(self.idp_keys) keys_to_check.append('domain_id') # create idp with the domain_id body = self.default_body.copy() body['description'] = uuid.uuid4().hex body['domain_id'] = domain['id'] idp1 = self._create_default_idp(body=body) self.assertValidResponse(idp1, 'identity_provider', dummy_validator, keys_to_check=keys_to_check, ref=body) # create a 2nd idp with the same domain_id url = self.base_url(suffix=uuid.uuid4().hex) body = self.default_body.copy() body['description'] = uuid.uuid4().hex body['domain_id'] = domain['id'] idp2 = self.put(url, body={'identity_provider': body}, expected_status=http.client.CREATED) self.assertValidResponse(idp2, 'identity_provider', dummy_validator, keys_to_check=keys_to_check, ref=body) self.assertEqual(idp1.result['identity_provider']['domain_id'], idp2.result['identity_provider']['domain_id']) def test_cannot_update_idp_domain(self): # create new idp body = self.default_body.copy() default_resp = self._create_default_idp(body=body) default_idp = self._fetch_attribute_from_response(default_resp, 'identity_provider') idp_id = default_idp.get('id') self.assertIsNotNone(idp_id) # create domain and try to update the idp's domain domain = unit.new_domain_ref() PROVIDERS.resource_api.create_domain(domain['id'], domain) body['domain_id'] = domain['id'] body = {'identity_provider': body} url = self.base_url(suffix=idp_id) self.patch(url, body=body, expected_status=http.client.BAD_REQUEST) def test_create_idp_with_nonexistent_domain_id_fails(self): body = self.default_body.copy() body['description'] = uuid.uuid4().hex body['domain_id'] = uuid.uuid4().hex self._create_default_idp(body=body, expected_status=http.client.NOT_FOUND) def test_create_idp_remote(self): """Create the IdentityProvider entity associated to remote_ids.""" keys_to_check = list(self.idp_keys) keys_to_check.append('remote_ids') body = self.default_body.copy() body['description'] = uuid.uuid4().hex body['remote_ids'] = [uuid.uuid4().hex, uuid.uuid4().hex, uuid.uuid4().hex] resp = self._create_default_idp(body=body) self.assertValidResponse(resp, 'identity_provider', dummy_validator, keys_to_check=keys_to_check, ref=body) attr = self._fetch_attribute_from_response(resp, 'identity_provider') self.assertIdpDomainCreated(attr['id'], attr['domain_id']) def test_create_idp_remote_repeated(self): """Create two IdentityProvider entities with some remote_ids. A remote_id is the same for both so the second IdP is not created because of the uniqueness of the remote_ids Expect HTTP 409 Conflict code for the latter call. """ body = self.default_body.copy() repeated_remote_id = uuid.uuid4().hex body['remote_ids'] = [uuid.uuid4().hex, uuid.uuid4().hex, uuid.uuid4().hex, repeated_remote_id] self._create_default_idp(body=body) url = self.base_url(suffix=uuid.uuid4().hex) body['remote_ids'] = [uuid.uuid4().hex, repeated_remote_id] resp = self.put(url, body={'identity_provider': body}, expected_status=http.client.CONFLICT) resp_data = jsonutils.loads(resp.body) self.assertIn('Duplicate remote ID', resp_data.get('error', {}).get('message')) def test_create_idp_remote_empty(self): """Create an IdP with empty remote_ids.""" keys_to_check = list(self.idp_keys) keys_to_check.append('remote_ids') body = self.default_body.copy() body['description'] = uuid.uuid4().hex body['remote_ids'] = [] resp = self._create_default_idp(body=body) self.assertValidResponse(resp, 'identity_provider', dummy_validator, keys_to_check=keys_to_check, ref=body) def test_create_idp_remote_none(self): """Create an IdP with a None remote_ids.""" keys_to_check = list(self.idp_keys) keys_to_check.append('remote_ids') body = self.default_body.copy() body['description'] = uuid.uuid4().hex body['remote_ids'] = None resp = self._create_default_idp(body=body) expected = body.copy() expected['remote_ids'] = [] self.assertValidResponse(resp, 'identity_provider', dummy_validator, keys_to_check=keys_to_check, ref=expected) def test_create_idp_authorization_ttl(self): keys_to_check = list(self.idp_keys) keys_to_check.append('authorization_ttl') body = self.default_body.copy() body['description'] = uuid.uuid4().hex body['authorization_ttl'] = 10080 resp = self._create_default_idp(body) expected = body.copy() self.assertValidResponse(resp, 'identity_provider', dummy_validator, keys_to_check=keys_to_check, ref=expected) def test_update_idp_remote_ids(self): """Update IdP's remote_ids parameter.""" body = self.default_body.copy() body['remote_ids'] = [uuid.uuid4().hex] default_resp = self._create_default_idp(body=body) default_idp = self._fetch_attribute_from_response(default_resp, 'identity_provider') idp_id = default_idp.get('id') url = self.base_url(suffix=idp_id) self.assertIsNotNone(idp_id) body['remote_ids'] = [uuid.uuid4().hex, uuid.uuid4().hex] body = {'identity_provider': body} resp = self.patch(url, body=body) updated_idp = self._fetch_attribute_from_response(resp, 'identity_provider') body = body['identity_provider'] self.assertEqual(sorted(body['remote_ids']), sorted(updated_idp.get('remote_ids'))) resp = self.get(url) returned_idp = self._fetch_attribute_from_response(resp, 'identity_provider') self.assertEqual(sorted(body['remote_ids']), sorted(returned_idp.get('remote_ids'))) def test_update_idp_clean_remote_ids(self): """Update IdP's remote_ids parameter with an empty list.""" body = self.default_body.copy() body['remote_ids'] = [uuid.uuid4().hex] default_resp = self._create_default_idp(body=body) default_idp = self._fetch_attribute_from_response(default_resp, 'identity_provider') idp_id = default_idp.get('id') url = self.base_url(suffix=idp_id) self.assertIsNotNone(idp_id) body['remote_ids'] = [] body = {'identity_provider': body} resp = self.patch(url, body=body) updated_idp = self._fetch_attribute_from_response(resp, 'identity_provider') body = body['identity_provider'] self.assertEqual(sorted(body['remote_ids']), sorted(updated_idp.get('remote_ids'))) resp = self.get(url) returned_idp = self._fetch_attribute_from_response(resp, 'identity_provider') self.assertEqual(sorted(body['remote_ids']), sorted(returned_idp.get('remote_ids'))) def test_update_idp_remote_repeated(self): """Update an IdentityProvider entity reusing a remote_id. A remote_id is the same for both so the second IdP is not updated because of the uniqueness of the remote_ids. Expect HTTP 409 Conflict code for the latter call. """ # Create first identity provider body = self.default_body.copy() repeated_remote_id = uuid.uuid4().hex body['remote_ids'] = [uuid.uuid4().hex, repeated_remote_id] self._create_default_idp(body=body) # Create second identity provider (without remote_ids) body = self.default_body.copy() default_resp = self._create_default_idp(body=body) default_idp = self._fetch_attribute_from_response(default_resp, 'identity_provider') idp_id = default_idp.get('id') url = self.base_url(suffix=idp_id) body['remote_ids'] = [repeated_remote_id] resp = self.patch(url, body={'identity_provider': body}, expected_status=http.client.CONFLICT) resp_data = jsonutils.loads(resp.body) self.assertIn('Duplicate remote ID', resp_data['error']['message']) def test_update_idp_authorization_ttl(self): body = self.default_body.copy() body['authorization_ttl'] = 10080 default_resp = self._create_default_idp(body=body) default_idp = self._fetch_attribute_from_response(default_resp, 'identity_provider') idp_id = default_idp.get('id') url = self.base_url(suffix=idp_id) self.assertIsNotNone(idp_id) body['authorization_ttl'] = None body = {'identity_provider': body} resp = self.patch(url, body=body) updated_idp = self._fetch_attribute_from_response(resp, 'identity_provider') body = body['identity_provider'] self.assertEqual(body['authorization_ttl'], updated_idp.get('authorization_ttl')) resp = self.get(url) returned_idp = self._fetch_attribute_from_response(resp, 'identity_provider') self.assertEqual(body['authorization_ttl'], returned_idp.get('authorization_ttl')) def test_list_head_idps(self, iterations=5): """List all available IdentityProviders. This test collects ids of created IdPs and intersects it with the list of all available IdPs. List of all IdPs can be a superset of IdPs created in this test, because other tests also create IdPs. """ def get_id(resp): r = self._fetch_attribute_from_response(resp, 'identity_provider') return r.get('id') ids = [] for _ in range(iterations): id = get_id(self._create_default_idp()) ids.append(id) ids = set(ids) keys_to_check = self.idp_keys keys_to_check.append('domain_id') url = self.base_url() resp = self.get(url) self.assertValidListResponse(resp, 'identity_providers', dummy_validator, keys_to_check=keys_to_check) entities = self._fetch_attribute_from_response(resp, 'identity_providers') entities_ids = set([e['id'] for e in entities]) ids_intersection = entities_ids.intersection(ids) self.assertEqual(ids_intersection, ids) self.head(url, expected_status=http.client.OK) def test_filter_list_head_idp_by_id(self): def get_id(resp): r = self._fetch_attribute_from_response(resp, 'identity_provider') return r.get('id') idp1_id = get_id(self._create_default_idp()) idp2_id = get_id(self._create_default_idp()) # list the IdP, should get two IdP. url = self.base_url() resp = self.get(url) entities = self._fetch_attribute_from_response(resp, 'identity_providers') entities_ids = [e['id'] for e in entities] self.assertCountEqual(entities_ids, [idp1_id, idp2_id]) # filter the IdP by ID. url = self.base_url() + '?id=' + idp1_id resp = self.get(url) filtered_service_list = resp.json['identity_providers'] self.assertThat(filtered_service_list, matchers.HasLength(1)) self.assertEqual(idp1_id, filtered_service_list[0].get('id')) self.head(url, expected_status=http.client.OK) def test_filter_list_head_idp_by_enabled(self): def get_id(resp): r = self._fetch_attribute_from_response(resp, 'identity_provider') return r.get('id') idp1_id = get_id(self._create_default_idp()) body = self.default_body.copy() body['enabled'] = False idp2_id = get_id(self._create_default_idp(body=body)) # list the IdP, should get two IdP. url = self.base_url() resp = self.get(url) entities = self._fetch_attribute_from_response(resp, 'identity_providers') entities_ids = [e['id'] for e in entities] self.assertCountEqual(entities_ids, [idp1_id, idp2_id]) # filter the IdP by 'enabled'. url = self.base_url() + '?enabled=True' resp = self.get(url) filtered_service_list = resp.json['identity_providers'] self.assertThat(filtered_service_list, matchers.HasLength(1)) self.assertEqual(idp1_id, filtered_service_list[0].get('id')) self.head(url, expected_status=http.client.OK) def test_check_idp_uniqueness(self): """Add same IdP twice. Expect HTTP 409 Conflict code for the latter call. """ url = self.base_url(suffix=uuid.uuid4().hex) body = self._http_idp_input() domain = unit.new_domain_ref() PROVIDERS.resource_api.create_domain(domain['id'], domain) body['domain_id'] = domain['id'] self.put(url, body={'identity_provider': body}, expected_status=http.client.CREATED) resp = self.put(url, body={'identity_provider': body}, expected_status=http.client.CONFLICT) resp_data = jsonutils.loads(resp.body) self.assertIn('Duplicate entry', resp_data.get('error', {}).get('message')) def test_get_head_idp(self): """Create and later fetch IdP.""" body = self._http_idp_input() domain = unit.new_domain_ref() PROVIDERS.resource_api.create_domain(domain['id'], domain) body['domain_id'] = domain['id'] default_resp = self._create_default_idp(body=body) default_idp = self._fetch_attribute_from_response(default_resp, 'identity_provider') idp_id = default_idp.get('id') url = self.base_url(suffix=idp_id) resp = self.get(url) # Strip keys out of `body` dictionary. This is done # to be python 3 compatible body_keys = list(body) self.assertValidResponse(resp, 'identity_provider', dummy_validator, keys_to_check=body_keys, ref=body) self.head(url, expected_status=http.client.OK) def test_get_nonexisting_idp(self): """Fetch nonexisting IdP entity. Expected HTTP 404 Not Found status code. """ idp_id = uuid.uuid4().hex self.assertIsNotNone(idp_id) url = self.base_url(suffix=idp_id) self.get(url, expected_status=http.client.NOT_FOUND) def test_delete_existing_idp(self): """Create and later delete IdP. Expect HTTP 404 Not Found for the GET IdP call. """ default_resp = self._create_default_idp() default_idp = self._fetch_attribute_from_response(default_resp, 'identity_provider') idp_id = default_idp.get('id') self.assertIsNotNone(idp_id) url = self.base_url(suffix=idp_id) self.delete(url) self.get(url, expected_status=http.client.NOT_FOUND) def test_delete_idp_also_deletes_assigned_protocols(self): """Deleting an IdP will delete its assigned protocol.""" # create default IdP default_resp = self._create_default_idp() default_idp = self._fetch_attribute_from_response(default_resp, 'identity_provider') idp_id = default_idp['id'] protocol_id = uuid.uuid4().hex url = self.base_url(suffix='%(idp_id)s/protocols/%(protocol_id)s') idp_url = self.base_url(suffix=idp_id) # assign protocol to IdP kwargs = {'expected_status': http.client.CREATED} resp, idp_id, proto = self._assign_protocol_to_idp( url=url, idp_id=idp_id, proto=protocol_id, **kwargs) # removing IdP will remove the assigned protocol as well self.assertEqual( 1, len(PROVIDERS.federation_api.list_protocols(idp_id)) ) self.delete(idp_url) self.get(idp_url, expected_status=http.client.NOT_FOUND) self.assertEqual( 0, len(PROVIDERS.federation_api.list_protocols(idp_id)) ) def test_delete_nonexisting_idp(self): """Delete nonexisting IdP. Expect HTTP 404 Not Found for the GET IdP call. """ idp_id = uuid.uuid4().hex url = self.base_url(suffix=idp_id) self.delete(url, expected_status=http.client.NOT_FOUND) def test_update_idp_mutable_attributes(self): """Update IdP's mutable parameters.""" default_resp = self._create_default_idp() default_idp = self._fetch_attribute_from_response(default_resp, 'identity_provider') idp_id = default_idp.get('id') url = self.base_url(suffix=idp_id) self.assertIsNotNone(idp_id) _enabled = not default_idp.get('enabled') body = {'remote_ids': [uuid.uuid4().hex, uuid.uuid4().hex], 'description': uuid.uuid4().hex, 'enabled': _enabled} body = {'identity_provider': body} resp = self.patch(url, body=body) def test_create_mapping_with_blacklist_and_whitelist(self): """Test for adding whitelist and blacklist in the rule Server should respond with HTTP 400 error upon discovering both ``whitelist`` and ``blacklist`` keywords in the same rule. """ url = self.MAPPING_URL + uuid.uuid4().hex self.assertEqual(body[key], updated_idp.get(key)) resp = self.get(url) updated_idp = self._fetch_attribute_from_response(resp, 'identity_provider') for key in body.keys(): if isinstance(body[key], list): self.assertEqual(sorted(body[key]), sorted(updated_idp.get(key))) else: self.assertEqual(body[key], updated_idp.get(key)) def test_update_idp_immutable_attributes(self): """Update IdP's immutable parameters. Expect HTTP BAD REQUEST. """ default_resp = self._create_default_idp() default_idp = self._fetch_attribute_from_response(default_resp, 'identity_provider') idp_id = default_idp.get('id') self.assertIsNotNone(idp_id) body = self._http_idp_input() body['id'] = uuid.uuid4().hex body['protocols'] = [uuid.uuid4().hex, uuid.uuid4().hex] url = self.base_url(suffix=idp_id) self.patch(url, body={'identity_provider': body}, expected_status=http.client.BAD_REQUEST) def test_update_nonexistent_idp(self): """Update nonexistent IdP. Expect HTTP 404 Not Found code. """ idp_id = uuid.uuid4().hex url = self.base_url(suffix=idp_id) body = self._http_idp_input() body['enabled'] = False body = {'identity_provider': body} self.patch(url, body=body, expected_status=http.client.NOT_FOUND) def test_assign_protocol_to_idp(self): """Assign a protocol to existing IdP.""" self._assign_protocol_to_idp(expected_status=http.client.CREATED) def test_protocol_composite_pk(self): """Test that Keystone can add two entities. The entities have identical names, however, attached to different IdPs. 1. Add IdP and assign it protocol with predefined name 2. Add another IdP and assign it a protocol with same name. Expect HTTP 201 code """ url = self.base_url(suffix='%(idp_id)s/protocols/%(protocol_id)s') kwargs = {'expected_status': http.client.CREATED} self._assign_protocol_to_idp(proto='saml2', url=url, **kwargs) self._assign_protocol_to_idp(proto='saml2', url=url, **kwargs) def test_protocol_idp_pk_uniqueness(self): """Test whether Keystone checks for unique idp/protocol values. Add same protocol twice, expect Keystone to reject a latter call and return HTTP 409 Conflict code. """ url = self.base_url(suffix='%(idp_id)s/protocols/%(protocol_id)s') kwargs = {'expected_status': http.client.CREATED} resp, idp_id, proto = self._assign_protocol_to_idp(proto='saml2', url=url, **kwargs) kwargs = {'expected_status': http.client.CONFLICT} self._assign_protocol_to_idp( idp_id=idp_id, proto='saml2', validate=False, url=url, **kwargs ) def test_assign_protocol_to_nonexistent_idp(self): """Assign protocol to IdP that doesn't exist. Expect HTTP 404 Not Found code. """ idp_id = uuid.uuid4().hex kwargs = {'expected_status': http.client.NOT_FOUND} self._assign_protocol_to_idp(proto='saml2', idp_id=idp_id, validate=False, **kwargs) def test_crud_protocol_without_protocol_id_in_url(self): # NOTE(morgan): This test is redundant but is added to ensure # the url routing error in bug 1817313 is explicitly covered. # create a protocol, but do not put the ID in the URL idp_id, _ = self._create_and_decapsulate_response() mapping_id = uuid.uuid4().hex self._create_mapping(mapping_id=mapping_id) protocol = { 'id': uuid.uuid4().hex, 'mapping_id': mapping_id } with self.test_client() as c: token = self.get_scoped_token() # DELETE/PATCH/PUT on non-trailing `/` results in # METHOD_NOT_ALLOWED c.delete('/v3/OS-FEDERATION/identity_providers/%(idp_id)s' '/protocols' % {'idp_id': idp_id}, headers={'X-Auth-Token': token}, expected_status_code=http.client.METHOD_NOT_ALLOWED) c.patch('/v3/OS-FEDERATION/identity_providers/%(idp_id)s' '/protocols/' % {'idp_id': idp_id}, json={'protocol': protocol}, headers={'X-Auth-Token': token}, expected_status_code=http.client.METHOD_NOT_ALLOWED) c.put('/v3/OS-FEDERATION/identity_providers/%(idp_id)s' '/protocols' % {'idp_id': idp_id}, json={'protocol': protocol}, headers={'X-Auth-Token': token}, expected_status_code=http.client.METHOD_NOT_ALLOWED) # DELETE/PATCH/PUT should raise 405 with trailing '/', it is # remapped to without the trailing '/' by the normalization # middleware. c.delete('/v3/OS-FEDERATION/identity_providers/%(idp_id)s' '/protocols/' % {'idp_id': idp_id}, headers={'X-Auth-Token': token}, expected_status_code=http.client.METHOD_NOT_ALLOWED) c.patch('/v3/OS-FEDERATION/identity_providers/%(idp_id)s' '/protocols/' % {'idp_id': idp_id}, json={'protocol': protocol}, headers={'X-Auth-Token': token}, expected_status_code=http.client.METHOD_NOT_ALLOWED) c.put('/v3/OS-FEDERATION/identity_providers/%(idp_id)s' '/protocols/' % {'idp_id': idp_id}, json={'protocol': protocol}, headers={'X-Auth-Token': token}, expected_status_code=http.client.METHOD_NOT_ALLOWED) def test_get_head_protocol(self): """Create and later fetch protocol tied to IdP.""" resp, idp_id, proto = self._assign_protocol_to_idp( expected_status=http.client.CREATED) proto_id = self._fetch_attribute_from_response(resp, 'protocol')['id'] url = "%s/protocols/%s" % (idp_id, proto_id) url = self.base_url(suffix=url) resp = self.get(url) reference = {'id': proto_id} # Strip keys out of `body` dictionary. This is done # to be python 3 compatible reference_keys = list(reference) self.assertValidResponse(resp, 'protocol', dummy_validator, keys_to_check=reference_keys, ref=reference) self.head(url, expected_status=http.client.OK) def test_list_head_protocols(self): """Create set of protocols and later list them. Compare input and output id sets. """ resp, idp_id, proto = self._assign_protocol_to_idp( expected_status=http.client.CREATED) iterations = random.randint(0, 16) protocol_ids = [] for _ in range(iterations): resp, _, proto = self._assign_protocol_to_idp( idp_id=idp_id, expected_status=http.client.CREATED) proto_id = self._fetch_attribute_from_response(resp, 'protocol') proto_id = proto_id['id'] protocol_ids.append(proto_id) url = "%s/protocols" % idp_id url = self.base_url(suffix=url) resp = self.get(url) self.assertValidListResponse(resp, 'protocols', dummy_validator, keys_to_check=['id']) entities = self._fetch_attribute_from_response(resp, 'protocols') entities = set([entity['id'] for entity in entities]) protocols_intersection = entities.intersection(protocol_ids) self.assertEqual(protocols_intersection, set(protocol_ids)) self.head(url, expected_status=http.client.OK) def test_update_protocols_attribute(self): """Update protocol's attribute.""" resp, idp_id, proto = self._assign_protocol_to_idp( expected_status=http.client.CREATED) new_mapping_id = uuid.uuid4().hex self._create_mapping(mapping_id=new_mapping_id) url = "%s/protocols/%s" % (idp_id, proto) url = self.base_url(suffix=url) body = {'mapping_id': new_mapping_id} resp = self.patch(url, body={'protocol': body}) self.assertValidResponse(resp, 'protocol', dummy_validator, keys_to_check=['id', 'mapping_id'], ref={'id': proto, 'mapping_id': new_mapping_id} ) def test_delete_protocol(self): """Delete protocol. Expect HTTP 404 Not Found code for the GET call after the protocol is deleted. """ url = self.base_url(suffix='%(idp_id)s/' 'protocols/%(protocol_id)s') resp, idp_id, proto = self._assign_protocol_to_idp( expected_status=http.client.CREATED) url = url % {'idp_id': idp_id, 'protocol_id': proto} self.delete(url) self.get(url, expected_status=http.client.NOT_FOUND) class MappingCRUDTests(test_v3.RestfulTestCase): """A class for testing CRUD operations for Mappings.""" MAPPING_URL = '/OS-FEDERATION/mappings/' def assertValidMappingListResponse(self, resp, *args, **kwargs): return self.assertValidListResponse( resp, 'mappings', self.assertValidMapping, keys_to_check=[], *args, **kwargs) def assertValidMappingResponse(self, resp, *args, **kwargs): return self.assertValidResponse( resp, 'mapping', self.assertValidMapping, keys_to_check=[], *args, **kwargs) def assertValidMapping(self, entity, ref=None): self.assertIsNotNone(entity.get('id')) self.assertIsNotNone(entity.get('rules')) if ref: self.assertEqual(entity['rules'], ref['rules']) return entity def _create_default_mapping_entry(self): url = self.MAPPING_URL + uuid.uuid4().hex resp = self.put(url, body={'mapping': mapping_fixtures.MAPPING_LARGE}, expected_status=http.client.CREATED) return resp def _get_id_from_response(self, resp): r = resp.result.get('mapping') return r.get('id') def test_mapping_create(self): resp = self._create_default_mapping_entry() self.assertValidMappingResponse(resp, mapping_fixtures.MAPPING_LARGE) def test_mapping_list_head(self): url = self.MAPPING_URL self._create_default_mapping_entry() resp = self.get(url) entities = resp.result.get('mappings') self.assertIsNotNone(entities) self.assertResponseStatus(resp, http.client.OK) self.assertValidListLinks(resp.result.get('links')) self.assertEqual(1, len(entities)) self.head(url, expected_status=http.client.OK) def test_mapping_delete(self): url = self.MAPPING_URL + '%(mapping_id)s' resp = self._create_default_mapping_entry() mapping_id = self._get_id_from_response(resp) url = url % {'mapping_id': str(mapping_id)} resp = self.delete(url) self.assertResponseStatus(resp, http.client.NO_CONTENT) self.get(url, expected_status=http.client.NOT_FOUND) def test_mapping_get_head(self): url = self.MAPPING_URL + '%(mapping_id)s' resp = self._create_default_mapping_entry() mapping_id = self._get_id_from_response(resp) url = url % {'mapping_id': mapping_id} resp = self.get(url) self.assertValidMappingResponse(resp, mapping_fixtures.MAPPING_LARGE) self.head(url, expected_status=http.client.OK) def test_mapping_update(self): url = self.MAPPING_URL + '%(mapping_id)s' resp = self._create_default_mapping_entry() mapping_id = self._get_id_from_response(resp) url = url % {'mapping_id': mapping_id} resp = self.patch(url, body={'mapping': mapping_fixtures.MAPPING_SMALL}) self.assertValidMappingResponse(resp, mapping_fixtures.MAPPING_SMALL) resp = self.get(url) self.assertValidMappingResponse(resp, mapping_fixtures.MAPPING_SMALL) def test_delete_mapping_dne(self): url = self.MAPPING_URL + uuid.uuid4().hex self.delete(url, expected_status=http.client.NOT_FOUND) def test_get_mapping_dne(self): url = self.MAPPING_URL + uuid.uuid4().hex self.get(url, expected_status=http.client.NOT_FOUND) def test_create_mapping_bad_requirements(self): url = self.MAPPING_URL + uuid.uuid4().hex self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': mapping_fixtures.MAPPING_BAD_REQ}) def test_create_mapping_no_rules(self): url = self.MAPPING_URL + uuid.uuid4().hex self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': mapping_fixtures.MAPPING_NO_RULES}) def test_create_mapping_no_remote_objects(self): url = self.MAPPING_URL + uuid.uuid4().hex self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': mapping_fixtures.MAPPING_NO_REMOTE}) def test_create_mapping_bad_value(self): url = self.MAPPING_URL + uuid.uuid4().hex self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': mapping_fixtures.MAPPING_BAD_VALUE}) def test_create_mapping_missing_local(self): url = self.MAPPING_URL + uuid.uuid4().hex self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': mapping_fixtures.MAPPING_MISSING_LOCAL}) def test_create_mapping_missing_type(self): url = self.MAPPING_URL + uuid.uuid4().hex self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': mapping_fixtures.MAPPING_MISSING_TYPE}) def test_create_mapping_wrong_type(self): url = self.MAPPING_URL + uuid.uuid4().hex self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': mapping_fixtures.MAPPING_WRONG_TYPE}) def test_create_mapping_extra_remote_properties_not_any_of(self): url = self.MAPPING_URL + uuid.uuid4().hex mapping = mapping_fixtures.MAPPING_EXTRA_REMOTE_PROPS_NOT_ANY_OF self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': mapping}) def test_create_mapping_extra_remote_properties_any_one_of(self): url = self.MAPPING_URL + uuid.uuid4().hex mapping = mapping_fixtures.MAPPING_EXTRA_REMOTE_PROPS_ANY_ONE_OF self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': mapping}) def test_create_mapping_extra_remote_properties_just_type(self): url = self.MAPPING_URL + uuid.uuid4().hex mapping = mapping_fixtures.MAPPING_EXTRA_REMOTE_PROPS_JUST_TYPE self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': mapping}) def test_create_mapping_empty_map(self): url = self.MAPPING_URL + uuid.uuid4().hex self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': {}}) def test_create_mapping_extra_rules_properties(self): url = self.MAPPING_URL + uuid.uuid4().hex self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': mapping_fixtures.MAPPING_EXTRA_RULES_PROPS}) def test_create_mapping_with_blacklist_and_whitelist(self): """Test for adding whitelist and blacklist in the rule. Server should respond with HTTP 400 Bad Request error upon discovering both ``whitelist`` and ``blacklist`` keywords in the same rule. """ url = self.MAPPING_URL + uuid.uuid4().hex mapping = mapping_fixtures.MAPPING_GROUPS_WHITELIST_AND_BLACKLIST self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': mapping}) def test_create_mapping_with_local_user_and_local_domain(self): url = self.MAPPING_URL + uuid.uuid4().hex resp = self.put( url, body={ 'mapping': mapping_fixtures.MAPPING_LOCAL_USER_LOCAL_DOMAIN }, expected_status=http.client.CREATED) self.assertValidMappingResponse( resp, mapping_fixtures.MAPPING_LOCAL_USER_LOCAL_DOMAIN) def test_create_mapping_with_ephemeral(self): url = self.MAPPING_URL + uuid.uuid4().hex resp = self.put( url, body={'mapping': mapping_fixtures.MAPPING_EPHEMERAL_USER}, expected_status=http.client.CREATED) self.assertValidMappingResponse( resp, mapping_fixtures.MAPPING_EPHEMERAL_USER) def test_create_mapping_with_bad_user_type(self): url = self.MAPPING_URL + uuid.uuid4().hex # get a copy of a known good map bad_mapping = copy.deepcopy(mapping_fixtures.MAPPING_EPHEMERAL_USER) # now sabotage the user type bad_mapping['rules'][0]['local'][0]['user']['type'] = uuid.uuid4().hex self.put(url, expected_status=http.client.BAD_REQUEST, body={'mapping': bad_mapping}) def test_create_shadow_mapping_without_roles_fails(self): """Validate that mappings with projects contain roles when created.""" url = self.MAPPING_URL + uuid.uuid4().hex self.put( url, body={'mapping': mapping_fixtures.MAPPING_PROJECTS_WITHOUT_ROLES}, expected_status=http.client.BAD_REQUEST ) def test_update_shadow_mapping_without_roles_fails(self): """Validate that mappings with projects contain roles when updated.""" url = self.MAPPING_URL + uuid.uuid4().hex resp = self.put( url, body={'mapping': mapping_fixtures.MAPPING_PROJECTS}, expected_status=http.client.CREATED ) self.assertValidMappingResponse( resp, mapping_fixtures.MAPPING_PROJECTS ) self.patch( url, body={'mapping': mapping_fixtures.MAPPING_PROJECTS_WITHOUT_ROLES}, expected_status=http.client.BAD_REQUEST ) def test_create_shadow_mapping_without_name_fails(self): """Validate project mappings contain the project name when created.""" url = self.MAPPING_URL + uuid.uuid4().hex self.put( url, body={'mapping': mapping_fixtures.MAPPING_PROJECTS_WITHOUT_NAME}, expected_status=http.client.BAD_REQUEST ) def test_update_shadow_mapping_without_name_fails(self): """Validate project mappings contain the project name when updated.""" url = self.MAPPING_URL + uuid.uuid4().hex resp = self.put( url, body={'mapping': mapping_fixtures.MAPPING_PROJECTS}, expected_status=http.client.CREATED ) self.assertValidMappingResponse( resp, mapping_fixtures.MAPPING_PROJECTS ) self.patch( url, body={'mapping': mapping_fixtures.MAPPING_PROJECTS_WITHOUT_NAME}, expected_status=http.client.BAD_REQUEST ) class FederatedTokenTests(test_v3.RestfulTestCase, FederatedSetupMixin): def auth_plugin_config_override(self): methods = ['saml2', 'token'] super(FederatedTokenTests, self).auth_plugin_config_override(methods) def setUp(self): super(FederatedTokenTests, self).setUp() self._notifications = [] def fake_saml_notify(action, user_id, group_ids, identity_provider, protocol, token_id, outcome): note = { 'action': action, 'user_id': user_id, 'identity_provider': identity_provider, 'protocol': protocol, 'send_notification_called': True} self._notifications.append(note) self.useFixture(fixtures.MockPatchObject( notifications, 'send_saml_audit_notification', fake_saml_notify)) def _assert_last_notify(self, action, identity_provider, protocol, user_id=None): self.assertTrue(self._notifications) note = self._notifications[-1] if user_id: self.assertEqual(note['user_id'], user_id) self.assertEqual(note['action'], action) self.assertEqual(note['identity_provider'], identity_provider) self.assertEqual(note['protocol'], protocol) self.assertTrue(note['send_notification_called']) def load_fixtures(self, fixtures): super(FederatedTokenTests, self).load_fixtures(fixtures) self.load_federation_sample_data() def test_issue_unscoped_token_notify(self): self._issue_unscoped_token() self._assert_last_notify(self.ACTION, self.IDP, self.PROTOCOL) def test_issue_unscoped_token(self): r = self._issue_unscoped_token() token_resp = render_token.render_token_response_from_model(r)['token'] self.assertValidMappedUser(token_resp) def test_default_domain_scoped_token(self): # Make sure federated users can get tokens scoped to the default # domain, which has a non-uuid ID by default (e.g., `default`). We want # to make sure the token provider handles string types properly if the # ID isn't compressed into byte format during validation. Turn off # cache on issue so that we validate the token online right after we # get it to make sure the token provider is called. self.config_fixture.config(group='token', cache_on_issue=False) # Grab an unscoped token to get a domain-scoped token with. token = self._issue_unscoped_token() # Give the user a direct role assignment on the default domain, so they # can get a federated domain-scoped token. PROVIDERS.assignment_api.create_grant( self.role_admin['id'], user_id=token.user_id, domain_id=CONF.identity.default_domain_id ) # Get a token scoped to the default domain with an ID of `default`, # which isn't a uuid type, but we should be able to handle it # accordingly in the token formatters/providers. auth_request = { 'auth': { 'identity': { 'methods': [ 'token' ], 'token': { 'id': token.id } }, 'scope': { 'domain': { 'id': CONF.identity.default_domain_id } } } } r = self.v3_create_token(auth_request) domain_scoped_token_id = r.headers.get('X-Subject-Token') # Validate the token to make sure the token providers handle non-uuid # domain IDs properly. headers = {'X-Subject-Token': domain_scoped_token_id} self.get( '/auth/tokens', token=domain_scoped_token_id, headers=headers ) def test_issue_the_same_unscoped_token_with_user_deleted(self): r = self._issue_unscoped_token() token = render_token.render_token_response_from_model(r)['token'] user1 = token['user'] user_id1 = user1.pop('id') # delete the referenced user, and authenticate again. Keystone should # create another new shadow user. PROVIDERS.identity_api.delete_user(user_id1) r = self._issue_unscoped_token() token = render_token.render_token_response_from_model(r)['token'] user2 = token['user'] user_id2 = user2.pop('id') # Only the user_id is different. Other properties include # identity_provider, protocol, groups and domain are the same. self.assertIsNot(user_id2, user_id1) self.assertEqual(user1, user2) def test_issue_unscoped_token_disabled_idp(self): """Check if authentication works with disabled identity providers. Test plan: 1) Disable default IdP 2) Try issuing unscoped token for that IdP 3) Expect server to forbid authentication """ enabled_false = {'enabled': False} PROVIDERS.federation_api.update_idp(self.IDP, enabled_false) self.assertRaises(exception.Forbidden, self._issue_unscoped_token) def test_issue_unscoped_token_group_names_in_mapping(self): r = self._issue_unscoped_token(assertion='ANOTHER_CUSTOMER_ASSERTION') ref_groups = set([self.group_customers['id'], self.group_admins['id']]) token_groups = r.federated_groups token_groups = set([group['id'] for group in token_groups]) self.assertEqual(ref_groups, token_groups) def test_issue_unscoped_tokens_nonexisting_group(self): self._issue_unscoped_token(assertion='ANOTHER_TESTER_ASSERTION') def test_issue_unscoped_token_with_remote_no_attribute(self): self._issue_unscoped_token(idp=self.IDP_WITH_REMOTE, environment={ self.REMOTE_ID_ATTR: self.REMOTE_IDS[0] }) def test_issue_unscoped_token_with_remote(self): self.config_fixture.config(group='federation', remote_id_attribute=self.REMOTE_ID_ATTR) self._issue_unscoped_token(idp=self.IDP_WITH_REMOTE, environment={ self.REMOTE_ID_ATTR: self.REMOTE_IDS[0] }) def test_issue_unscoped_token_with_saml2_remote(self): self.config_fixture.config(group='saml2', remote_id_attribute=self.REMOTE_ID_ATTR) self._issue_unscoped_token(idp=self.IDP_WITH_REMOTE, environment={ self.REMOTE_ID_ATTR: self.REMOTE_IDS[0] }) def test_issue_unscoped_token_with_remote_different(self): self.config_fixture.config(group='federation', remote_id_attribute=self.REMOTE_ID_ATTR) self.assertRaises(exception.Forbidden, self._issue_unscoped_token, idp=self.IDP_WITH_REMOTE, environment={ self.REMOTE_ID_ATTR: uuid.uuid4().hex }) def test_issue_unscoped_token_with_remote_default_overwritten(self): """Test that protocol remote_id_attribute has higher priority. Make sure the parameter stored under ``protocol`` section has higher priority over parameter from default ``federation`` configuration section. """ self.config_fixture.config(group='saml2', remote_id_attribute=self.REMOTE_ID_ATTR) self.config_fixture.config(group='federation', remote_id_attribute=uuid.uuid4().hex) self._issue_unscoped_token(idp=self.IDP_WITH_REMOTE, environment={ self.REMOTE_ID_ATTR: self.REMOTE_IDS[0] }) def test_issue_unscoped_token_with_remote_unavailable(self): self.config_fixture.config(group='federation', remote_id_attribute=self.REMOTE_ID_ATTR) self.assertRaises(exception.Unauthorized, self._issue_unscoped_token, idp=self.IDP_WITH_REMOTE, environment={ uuid.uuid4().hex: uuid.uuid4().hex }) def test_issue_unscoped_token_with_remote_user_as_empty_string(self): # make sure that REMOTE_USER set as the empty string won't interfere self._issue_unscoped_token(environment={'REMOTE_USER': ''}) def test_issue_unscoped_token_no_groups(self): r = self._issue_unscoped_token(assertion='USER_NO_GROUPS_ASSERTION') token_groups = r.federated_groups self.assertEqual(0, len(token_groups)) def test_issue_scoped_token_no_groups(self): """Verify that token without groups cannot get scoped to project. This test is required because of bug 1677723. """ # issue unscoped token with no groups r = self._issue_unscoped_token(assertion='USER_NO_GROUPS_ASSERTION') token_groups = r.federated_groups self.assertEqual(0, len(token_groups)) unscoped_token = r.id # let admin get roles in a project self.proj_employees admin = unit.new_user_ref(CONF.identity.default_domain_id) PROVIDERS.identity_api.create_user(admin) PROVIDERS.assignment_api.create_grant( self.role_admin['id'], user_id=admin['id'], project_id=self.proj_employees['id'] ) # try to scope the token. It should fail scope = self._scope_request( unscoped_token, 'project', self.proj_employees['id'] ) self.v3_create_token( scope, expected_status=http.client.UNAUTHORIZED) def test_issue_unscoped_token_malformed_environment(self): """Test whether non string objects are filtered out. Put non string objects into the environment, inject correct assertion and try to get an unscoped token. Expect server not to fail on using split() method on non string objects and return token id in the HTTP header. """ environ = { 'malformed_object': object(), 'another_bad_idea': tuple(range(10)), 'yet_another_bad_param': dict(zip(uuid.uuid4().hex, range(32))) } environ.update(mapping_fixtures.EMPLOYEE_ASSERTION) with self.make_request(environ=environ): authentication.authenticate_for_token(self.UNSCOPED_V3_SAML2_REQ) def test_scope_to_project_once_notify(self): r = self.v3_create_token( self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_EMPLOYEE) user_id = r.json['token']['user']['id'] self._assert_last_notify(self.ACTION, self.IDP, self.PROTOCOL, user_id) def test_scope_to_project_once(self): r = self.v3_create_token( self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_EMPLOYEE) token_resp = r.result['token'] project_id = token_resp['project']['id'] self._check_project_scoped_token_attributes(token_resp, project_id) roles_ref = [self.role_employee] projects_ref = self.proj_employees self._check_projects_and_roles(token_resp, roles_ref, projects_ref) self.assertValidMappedUser(token_resp) def test_scope_token_with_idp_disabled(self): """Scope token issued by disabled IdP. Try scoping the token issued by an IdP which is disabled now. Expect server to refuse scoping operation. This test confirms correct behaviour when IdP was enabled and unscoped token was issued, but disabled before user tries to scope the token. Here we assume the unscoped token was already issued and start from the moment where IdP is being disabled and unscoped token is being used. Test plan: 1) Disable IdP 2) Try scoping unscoped token """ enabled_false = {'enabled': False} PROVIDERS.federation_api.update_idp(self.IDP, enabled_false) self.v3_create_token( self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_CUSTOMER, expected_status=http.client.FORBIDDEN) def test_validate_token_after_deleting_idp_raises_not_found(self): token = self.v3_create_token( self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN ) token_id = token.headers.get('X-Subject-Token') federated_info = token.json_body['token']['user']['OS-FEDERATION'] idp_id = federated_info['identity_provider']['id'] PROVIDERS.federation_api.delete_idp(idp_id) headers = { 'X-Subject-Token': token_id } # NOTE(lbragstad): This raises a 404 NOT FOUND because the identity # provider is no longer present. We raise 404 NOT FOUND when we # validate a token and a project or domain no longer exists. self.get( '/auth/tokens/', token=token_id, headers=headers, expected_status=http.client.NOT_FOUND ) def test_deleting_idp_cascade_deleting_fed_user(self): token = self.v3_create_token( self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN ) federated_info = token.json_body['token']['user']['OS-FEDERATION'] idp_id = federated_info['identity_provider']['id'] # There are three fed users (from 'EMPLOYEE_ASSERTION', # 'CUSTOMER_ASSERTION', 'ADMIN_ASSERTION') with the specified idp. hints = driver_hints.Hints() hints.add_filter('idp_id', idp_id) fed_users = PROVIDERS.shadow_users_api.get_federated_users(hints) self.assertEqual(3, len(fed_users)) idp_domain_id = PROVIDERS.federation_api.get_idp(idp_id)['domain_id'] for fed_user in fed_users: self.assertEqual(idp_domain_id, fed_user['domain_id']) # Delete the idp PROVIDERS.federation_api.delete_idp(idp_id) # The related federated user should be deleted as well. hints = driver_hints.Hints() hints.add_filter('idp_id', idp_id) fed_users = PROVIDERS.shadow_users_api.get_federated_users(hints) self.assertEqual([], fed_users) def test_scope_to_bad_project(self): """Scope unscoped token with a project we don't have access to.""" self.v3_create_token( self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_CUSTOMER, expected_status=http.client.UNAUTHORIZED) def test_scope_to_project_multiple_times(self): """Try to scope the unscoped token multiple times. The new tokens should be scoped to: * Customers' project * Employees' project """ bodies = (self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN, self.TOKEN_SCOPE_PROJECT_CUSTOMER_FROM_ADMIN) project_ids = (self.proj_employees['id'], self.proj_customers['id']) for body, project_id_ref in zip(bodies, project_ids): r = self.v3_create_token(body) token_resp = r.result['token'] self._check_project_scoped_token_attributes(token_resp, project_id_ref) def test_scope_to_project_with_duplicate_roles_returns_single_role(self): r = self.v3_create_token(self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN) # Even though the process of obtaining a token shows that there is a # role assignment on a project, we should attempt to create a duplicate # assignment somewhere. Do this by creating a direct role assignment # with each role against the project the token was scoped to. user_id = r.json_body['token']['user']['id'] project_id = r.json_body['token']['project']['id'] for role in r.json_body['token']['roles']: PROVIDERS.assignment_api.create_grant( role_id=role['id'], user_id=user_id, project_id=project_id ) # Ensure all roles in the token are unique even though we know there # should be duplicate role assignment from the assertions and the # direct role assignments we just created. r = self.v3_create_token(self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN) known_role_ids = [] for role in r.json_body['token']['roles']: self.assertNotIn(role['id'], known_role_ids) known_role_ids.append(role['id']) def test_scope_to_project_with_only_inherited_roles(self): """Try to scope token whose only roles are inherited.""" r = self.v3_create_token( self.TOKEN_SCOPE_PROJECT_INHERITED_FROM_CUSTOMER) token_resp = r.result['token'] self._check_project_scoped_token_attributes( token_resp, self.project_inherited['id']) roles_ref = [self.role_customer] projects_ref = self.project_inherited self._check_projects_and_roles(token_resp, roles_ref, projects_ref) self.assertValidMappedUser(token_resp) def test_scope_token_from_nonexistent_unscoped_token(self): """Try to scope token from non-existent unscoped token.""" self.v3_create_token( self.TOKEN_SCOPE_PROJECT_FROM_NONEXISTENT_TOKEN, expected_status=http.client.NOT_FOUND) def test_issue_token_from_rules_without_user(self): environ = copy.deepcopy(mapping_fixtures.BAD_TESTER_ASSERTION) with self.make_request(environ=environ): self.assertRaises(exception.Unauthorized, authentication.authenticate_for_token, self.UNSCOPED_V3_SAML2_REQ) def test_issue_token_with_nonexistent_group(self): """Inject assertion that matches rule issuing bad group id. Expect server to find out that some groups are missing in the backend and raise exception.MappedGroupNotFound exception. """ self.assertRaises(exception.MappedGroupNotFound, self._issue_unscoped_token, assertion='CONTRACTOR_ASSERTION') def test_scope_to_domain_once(self): r = self.v3_create_token(self.TOKEN_SCOPE_DOMAIN_A_FROM_CUSTOMER) token_resp = r.result['token'] self._check_domain_scoped_token_attributes(token_resp, self.domainA['id']) def test_scope_to_domain_multiple_tokens(self): """Issue multiple tokens scoping to different domains. The new tokens should be scoped to: * domainA * domainB * domainC """ bodies = (self.TOKEN_SCOPE_DOMAIN_A_FROM_ADMIN, self.TOKEN_SCOPE_DOMAIN_B_FROM_ADMIN, self.TOKEN_SCOPE_DOMAIN_C_FROM_ADMIN) domain_ids = (self.domainA['id'], self.domainB['id'], self.domainC['id']) for body, domain_id_ref in zip(bodies, domain_ids): r = self.v3_create_token(body) token_resp = r.result['token'] self._check_domain_scoped_token_attributes(token_resp, domain_id_ref) def test_scope_to_domain_with_only_inherited_roles_fails(self): """Try to scope to a domain that has no direct roles.""" self.v3_create_token( self.TOKEN_SCOPE_DOMAIN_D_FROM_CUSTOMER, expected_status=http.client.UNAUTHORIZED) def test_list_projects(self): urls = ('/OS-FEDERATION/projects', '/auth/projects') token = (self.tokens['CUSTOMER_ASSERTION'], self.tokens['EMPLOYEE_ASSERTION'], self.tokens['ADMIN_ASSERTION']) projects_refs = (set([self.proj_customers['id'], self.project_inherited['id']]), set([self.proj_employees['id'], self.project_all['id']]), set([self.proj_employees['id'], self.project_all['id'], self.proj_customers['id'], self.project_inherited['id']])) for token, projects_ref in zip(token, projects_refs): for url in urls: r = self.get(url, token=token) projects_resp = r.result['projects'] projects = set(p['id'] for p in projects_resp) self.assertEqual(projects_ref, projects, 'match failed for url %s' % url) # TODO(samueldmq): Create another test class for role inheritance tests. # The advantage would be to reduce the complexity of this test class and # have tests specific to this functionality grouped, easing readability and # maintenability. def test_list_projects_for_inherited_project_assignment(self): # Create a subproject subproject_inherited = unit.new_project_ref( domain_id=self.domainD['id'], parent_id=self.project_inherited['id']) PROVIDERS.resource_api.create_project( subproject_inherited['id'], subproject_inherited ) # Create an inherited role assignment PROVIDERS.assignment_api.create_grant( role_id=self.role_employee['id'], group_id=self.group_employees['id'], project_id=self.project_inherited['id'], inherited_to_projects=True) # Define expected projects from employee assertion, which contain # the created subproject expected_project_ids = [self.project_all['id'], self.proj_employees['id'], subproject_inherited['id']] # Assert expected projects for both available URLs for url in ('/OS-FEDERATION/projects', '/auth/projects'): r = self.get(url, token=self.tokens['EMPLOYEE_ASSERTION']) project_ids = [project['id'] for project in r.result['projects']] self.assertEqual(len(expected_project_ids), len(project_ids)) for expected_project_id in expected_project_ids: self.assertIn(expected_project_id, project_ids, 'Projects match failed for url %s' % url) def test_list_domains(self): urls = ('/OS-FEDERATION/domains', '/auth/domains') tokens = (self.tokens['CUSTOMER_ASSERTION'], self.tokens['EMPLOYEE_ASSERTION'], self.tokens['ADMIN_ASSERTION']) # NOTE(henry-nash): domain D does not appear in the expected results # since it only had inherited roles (which only apply to projects # within the domain) domain_refs = (set([self.domainA['id']]), set([self.domainA['id'], self.domainB['id']]), set([self.domainA['id'], self.domainB['id'], self.domainC['id']])) for token, domains_ref in zip(tokens, domain_refs): for url in urls: r = self.get(url, token=token) domains_resp = r.result['domains'] domains = set(p['id'] for p in domains_resp) self.assertEqual(domains_ref, domains, 'match failed for url %s' % url) def test_full_workflow(self): """Test 'standard' workflow for granting access tokens. * Issue unscoped token * List available projects based on groups * Scope token to one of available projects """ r = self._issue_unscoped_token() token_resp = render_token.render_token_response_from_model(r)['token'] # NOTE(lbragstad): Ensure only 'saml2' is in the method list. self.assertListEqual(['saml2'], r.methods) self.assertValidMappedUser(token_resp) employee_unscoped_token_id = r.id r = self.get('/auth/projects', token=employee_unscoped_token_id) projects = r.result['projects'] random_project = random.randint(0, len(projects) - 1) project = projects[random_project] v3_scope_request = self._scope_request(employee_unscoped_token_id, 'project', project['id']) r = self.v3_create_token(v3_scope_request) token_resp = r.result['token'] self.assertIn('token', token_resp['methods']) self.assertIn('saml2', token_resp['methods']) self._check_project_scoped_token_attributes(token_resp, project['id']) def test_workflow_with_groups_deletion(self): """Test full workflow with groups deletion before token scoping. The test scenario is as follows: - Create group ``group`` - Create and assign roles to ``group`` and ``project_all`` - Patch mapping rules for existing IdP so it issues group id - Issue unscoped token with ``group``'s id - Delete group ``group`` - Scope token to ``project_all`` - Expect HTTP 500 response """ # create group and role group = unit.new_group_ref(domain_id=self.domainA['id']) group = PROVIDERS.identity_api.create_group(group) role = unit.new_role_ref() PROVIDERS.role_api.create_role(role['id'], role) # assign role to group and project_admins PROVIDERS.assignment_api.create_grant( role['id'], group_id=group['id'], project_id=self.project_all['id'] ) rules = { 'rules': [ { 'local': [ { 'group': { 'id': group['id'] } }, { 'user': { 'name': '{0}' } } ], 'remote': [ { 'type': 'UserName' }, { 'type': 'LastName', 'any_one_of': [ 'Account' ] } ] } ] } PROVIDERS.federation_api.update_mapping(self.mapping['id'], rules) r = self._issue_unscoped_token(assertion='TESTER_ASSERTION') # delete group PROVIDERS.identity_api.delete_group(group['id']) # scope token to project_all, expect HTTP 500 scoped_token = self._scope_request( r.id, 'project', self.project_all['id']) self.v3_create_token( scoped_token, expected_status=http.client.INTERNAL_SERVER_ERROR) def test_lists_with_missing_group_in_backend(self): """Test a mapping that points to a group that does not exist. For explicit mappings, we expect the group to exist in the backend, but for lists, specifically blacklists, a missing group is expected as many groups will be specified by the IdP that are not Keystone groups. The test scenario is as follows: - Create group ``EXISTS`` - Set mapping rules for existing IdP with a blacklist that passes through as REMOTE_USER_GROUPS - Issue unscoped token with on group ``EXISTS`` id in it """ domain_id = self.domainA['id'] domain_name = self.domainA['name'] group = unit.new_group_ref(domain_id=domain_id, name='EXISTS') group = PROVIDERS.identity_api.create_group(group) rules = { 'rules': [ { "local": [ { "user": { "name": "{0}", "id": "{0}" } } ], "remote": [ { "type": "REMOTE_USER" } ] }, { "local": [ { "groups": "{0}", "domain": {"name": domain_name} } ], "remote": [ { "type": "REMOTE_USER_GROUPS", } ] } ] } PROVIDERS.federation_api.update_mapping(self.mapping['id'], rules) r = self._issue_unscoped_token(assertion='UNMATCHED_GROUP_ASSERTION') assigned_group_ids = r.federated_groups self.assertEqual(1, len(assigned_group_ids)) self.assertEqual(group['id'], assigned_group_ids[0]['id']) def test_empty_blacklist_passess_all_values(self): """Test a mapping with empty blacklist specified. Not adding a ``blacklist`` keyword to the mapping rules has the same effect as adding an empty ``blacklist``. In both cases, the mapping engine will not discard any groups that are associated with apache environment variables. This test checks scenario where an empty blacklist was specified. Expected result is to allow any value. The test scenario is as follows: - Create group ``EXISTS`` - Create group ``NO_EXISTS`` - Set mapping rules for existing IdP with a blacklist that passes through as REMOTE_USER_GROUPS - Issue unscoped token with groups ``EXISTS`` and ``NO_EXISTS`` assigned """ domain_id = self.domainA['id'] domain_name = self.domainA['name'] # Add a group "EXISTS" group_exists = unit.new_group_ref(domain_id=domain_id, name='EXISTS') group_exists = PROVIDERS.identity_api.create_group(group_exists) # Add a group "NO_EXISTS" group_no_exists = unit.new_group_ref(domain_id=domain_id, name='NO_EXISTS') group_no_exists = PROVIDERS.identity_api.create_group(group_no_exists) group_ids = set([group_exists['id'], group_no_exists['id']]) rules = { 'rules': [ { "local": [ { "user": { "name": "{0}", "id": "{0}" } } ], "remote": [ { "type": "REMOTE_USER" } ] }, { "local": [ { "groups": "{0}", "domain": {"name": domain_name} } ], "remote": [ { "type": "REMOTE_USER_GROUPS", "blacklist": [] } ] } ] } PROVIDERS.federation_api.update_mapping(self.mapping['id'], rules) r = self._issue_unscoped_token(assertion='UNMATCHED_GROUP_ASSERTION') assigned_group_ids = r.federated_groups self.assertEqual(len(group_ids), len(assigned_group_ids)) for group in assigned_group_ids: self.assertIn(group['id'], group_ids) def test_not_adding_blacklist_passess_all_values(self): """Test a mapping without blacklist specified. Not adding a ``blacklist`` keyword to the mapping rules has the same effect as adding an empty ``blacklist``. In both cases all values will be accepted and passed. This test checks scenario where an blacklist was not specified. Expected result is to allow any value. The test scenario is as follows: - Create group ``EXISTS`` - Create group ``NO_EXISTS`` - Set mapping rules for existing IdP with a blacklist that passes through as REMOTE_USER_GROUPS - Issue unscoped token with on groups ``EXISTS`` and ``NO_EXISTS`` assigned """ domain_id = self.domainA['id'] domain_name = self.domainA['name'] # Add a group "EXISTS" group_exists = unit.new_group_ref(domain_id=domain_id, name='EXISTS') group_exists = PROVIDERS.identity_api.create_group(group_exists) # Add a group "NO_EXISTS" group_no_exists = unit.new_group_ref(domain_id=domain_id, name='NO_EXISTS') group_no_exists = PROVIDERS.identity_api.create_group(group_no_exists) group_ids = set([group_exists['id'], group_no_exists['id']]) rules = { 'rules': [ { "local": [ { "user": { "name": "{0}", "id": "{0}" } } ], "remote": [ { "type": "REMOTE_USER" } ] }, { def test_invalid_scope_body(self): """Test that missing the scope in request body raises an exception. Raises exception.SchemaValidationError() - error code 400 """ { "type": "REMOTE_USER_GROUPS", } ] } ] } PROVIDERS.federation_api.update_mapping(self.mapping['id'], rules) def test_invalid_token_body(self): """Test that missing the token in request body raises an exception. Raises exception.SchemaValidationError() - error code 400 """ """Test that empty whitelist blocks all the values. Not adding a ``whitelist`` keyword to the mapping value is different than adding empty whitelist. The former case will simply pass all the values, whereas the latter would discard all the values. This test checks scenario where an empty whitelist was specified. The expected result is that no groups are matched. The test scenario is as follows: - Create group ``EXISTS`` - Set mapping rules for existing IdP with an empty whitelist that whould discard any values from the assertion - Try issuing unscoped token, no groups were matched and that the federated user does not have any group assigned. """ domain_id = self.domainA['id'] domain_name = self.domainA['name'] group = unit.new_group_ref(domain_id=domain_id, name='EXISTS') group = PROVIDERS.identity_api.create_group(group) rules = { 'rules': [ { "local": [ { "user": { "name": "{0}", "id": "{0}" } } ], "remote": [ { "type": "REMOTE_USER" } ] }, { "local": [ { "groups": "{0}", "domain": {"name": domain_name} } ], "remote": [ { "type": "REMOTE_USER_GROUPS", "whitelist": [] } ] } ] } PROVIDERS.federation_api.update_mapping(self.mapping['id'], rules) r = self._issue_unscoped_token(assertion='UNMATCHED_GROUP_ASSERTION') assigned_groups = r.federated_groups self.assertEqual(len(assigned_groups), 0) def test_not_setting_whitelist_accepts_all_values(self): """Test that not setting whitelist passes. Not adding a ``whitelist`` keyword to the mapping value is different than adding empty whitelist. The former case will simply pass all the values, whereas the latter would discard all the values. This test checks a scenario where a ``whitelist`` was not specified. Expected result is that no groups are ignored. The test scenario is as follows: - Create group ``EXISTS`` - Set mapping rules for existing IdP with an empty whitelist that whould discard any values from the assertion - Issue an unscoped token and make sure ephemeral user is a member of two groups. """ domain_id = self.domainA['id'] domain_name = self.domainA['name'] # Add a group "EXISTS" group_exists = unit.new_group_ref(domain_id=domain_id, name='EXISTS') group_exists = PROVIDERS.identity_api.create_group(group_exists) # Add a group "NO_EXISTS" group_no_exists = unit.new_group_ref(domain_id=domain_id, name='NO_EXISTS') group_no_exists = PROVIDERS.identity_api.create_group(group_no_exists) group_ids = set([group_exists['id'], group_no_exists['id']]) rules = { 'rules': [ { "local": [ { "user": { "name": "{0}", "id": "{0}" } } ], "remote": [ { "type": "REMOTE_USER" } ] }, { "local": [ { "groups": "{0}", "domain": {"name": domain_name} } ], "remote": [ { "type": "REMOTE_USER_GROUPS", } ] } ] } PROVIDERS.federation_api.update_mapping(self.mapping['id'], rules) r = self._issue_unscoped_token(assertion='UNMATCHED_GROUP_ASSERTION') assigned_group_ids = r.federated_groups self.assertEqual(len(group_ids), len(assigned_group_ids)) for group in assigned_group_ids: self.assertIn(group['id'], group_ids) def test_assertion_prefix_parameter(self): """Test parameters filtering based on the prefix. With ``assertion_prefix`` set to fixed, non default value, issue an unscoped token from assertion EMPLOYEE_ASSERTION_PREFIXED. Expect server to return unscoped token. """ self.config_fixture.config(group='federation', assertion_prefix=self.ASSERTION_PREFIX) self._issue_unscoped_token(assertion='EMPLOYEE_ASSERTION_PREFIXED') def test_assertion_prefix_parameter_expect_fail(self): """Test parameters filtering based on the prefix. With ``assertion_prefix`` default value set to empty string issue an unscoped token from assertion EMPLOYEE_ASSERTION. Next, configure ``assertion_prefix`` to value ``UserName``. Try issuing unscoped token with EMPLOYEE_ASSERTION. Expect server to raise exception.Unathorized exception. """ self._issue_unscoped_token() self.config_fixture.config(group='federation', assertion_prefix='UserName') self.assertRaises(exception.Unauthorized, self._issue_unscoped_token) def test_unscoped_token_has_user_domain(self): r = self._issue_unscoped_token() self._check_domains_are_valid( render_token.render_token_response_from_model(r)['token']) def test_scoped_token_has_user_domain(self): r = self.v3_create_token( self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_EMPLOYEE) self._check_domains_are_valid(r.json_body['token']) def test_issue_unscoped_token_for_local_user(self): r = self._issue_unscoped_token(assertion='LOCAL_USER_ASSERTION') self.assertListEqual(['saml2'], r.methods) self.assertEqual(self.user['id'], r.user_id) self.assertEqual(self.user['name'], r.user['name']) self.assertEqual(self.domain['id'], r.user_domain['id']) # Make sure the token is not scoped self.assertIsNone(r.domain_id) self.assertIsNone(r.project_id) self.assertTrue(r.unscoped) def test_issue_token_for_local_user_user_not_found(self): self.assertRaises(exception.Unauthorized, self._issue_unscoped_token, assertion='ANOTHER_LOCAL_USER_ASSERTION') def test_user_name_and_id_in_federation_token(self): r = self._issue_unscoped_token(assertion='EMPLOYEE_ASSERTION') self.assertEqual( mapping_fixtures.EMPLOYEE_ASSERTION['UserName'], r.user['name']) self.assertNotEqual(r.user['name'], r.user_id) r = self.v3_create_token( self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_EMPLOYEE) token = r.json_body['token'] self.assertEqual( mapping_fixtures.EMPLOYEE_ASSERTION['UserName'], token['user']['name']) self.assertNotEqual(token['user']['name'], token['user']['id']) def test_issue_unscoped_token_with_remote_different_from_protocol(self): protocol = PROVIDERS.federation_api.get_protocol( self.IDP_WITH_REMOTE, self.PROTOCOL ) protocol['remote_id_attribute'] = uuid.uuid4().hex PROVIDERS.federation_api.update_protocol( self.IDP_WITH_REMOTE, protocol['id'], protocol ) self._issue_unscoped_token( idp=self.IDP_WITH_REMOTE, environment={ protocol['remote_id_attribute']: self.REMOTE_IDS[0] } ) self.assertRaises( exception.Unauthorized, self._issue_unscoped_token, idp=self.IDP_WITH_REMOTE, environment={uuid.uuid4().hex: self.REMOTE_IDS[0]} ) class FernetFederatedTokenTests(test_v3.RestfulTestCase, FederatedSetupMixin): AUTH_METHOD = 'token' def load_fixtures(self, fixtures): super(FernetFederatedTokenTests, self).load_fixtures(fixtures) self.load_federation_sample_data() def config_overrides(self): super(FernetFederatedTokenTests, self).config_overrides() self.config_fixture.config(group='token', provider='fernet') self.useFixture( ksfixtures.KeyRepository( self.config_fixture, 'fernet_tokens', CONF.fernet_tokens.max_active_keys ) ) def auth_plugin_config_override(self): methods = ['saml2', 'token', 'password'] super(FernetFederatedTokenTests, self).auth_plugin_config_override(methods) def test_federated_unscoped_token(self): resp = self._issue_unscoped_token() self.assertValidMappedUser( render_token.render_token_response_from_model(resp)['token']) def test_federated_unscoped_token_with_multiple_groups(self): assertion = 'ANOTHER_CUSTOMER_ASSERTION' resp = self._issue_unscoped_token(assertion=assertion) self.assertValidMappedUser( render_token.render_token_response_from_model(resp)['token']) def test_validate_federated_unscoped_token(self): resp = self._issue_unscoped_token() unscoped_token = resp.id # assert that the token we received is valid self.get('/auth/tokens/', headers={'X-Subject-Token': unscoped_token}) def test_fernet_full_workflow(self): """Test 'standard' workflow for granting Fernet access tokens. * Issue unscoped token * List available projects based on groups * Scope token to one of available projects """ resp = self._issue_unscoped_token() self.assertValidMappedUser( render_token.render_token_response_from_model(resp)['token']) unscoped_token = resp.id resp = self.get('/auth/projects', token=unscoped_token) projects = resp.result['projects'] random_project = random.randint(0, len(projects) - 1) project = projects[random_project] v3_scope_request = self._scope_request(unscoped_token, 'project', project['id']) resp = self.v3_create_token(v3_scope_request) token_resp = resp.result['token'] self._check_project_scoped_token_attributes(token_resp, project['id']) class JWSFederatedTokenTests(test_v3.RestfulTestCase, FederatedSetupMixin): AUTH_METHOD = 'token' def load_fixtures(self, fixtures): super(JWSFederatedTokenTests, self).load_fixtures(fixtures) self.load_federation_sample_data() def config_overrides(self): super(JWSFederatedTokenTests, self).config_overrides() self.config_fixture.config(group='token', provider='jws') self.useFixture(ksfixtures.JWSKeyRepository(self.config_fixture)) def auth_plugin_config_override(self): methods = ['saml2', 'token', 'password'] super(JWSFederatedTokenTests, self).auth_plugin_config_override(methods) def test_federated_unscoped_token(self): token_model = self._issue_unscoped_token() self.assertValidMappedUser( render_token.render_token_response_from_model(token_model)['token'] ) def test_federated_unscoped_token_with_multiple_groups(self): assertion = 'ANOTHER_CUSTOMER_ASSERTION' token_model = self._issue_unscoped_token(assertion=assertion) self.assertValidMappedUser( render_token.render_token_response_from_model(token_model)['token'] ) def test_validate_federated_unscoped_token(self): token_model = self._issue_unscoped_token() unscoped_token = token_model.id # assert that the token we received is valid self.get('/auth/tokens/', headers={'X-Subject-Token': unscoped_token}) def test_jws_full_workflow(self): """Test 'standard' workflow for granting JWS tokens. * Issue unscoped token * List available projects based on groups * Scope token to one of available projects """ token_model = self._issue_unscoped_token() self.assertValidMappedUser( render_token.render_token_response_from_model(token_model)['token'] ) unscoped_token = token_model.id resp = self.get('/auth/projects', token=unscoped_token) projects = resp.result['projects'] random_project = random.randint(0, len(projects) - 1) project = projects[random_project] v3_scope_request = self._scope_request(unscoped_token, 'project', project['id']) resp = self.v3_create_token(v3_scope_request) token_resp = resp.result['token'] self._check_project_scoped_token_attributes(token_resp, project['id']) class FederatedTokenTestsMethodToken(FederatedTokenTests): """Test federation operation with unified scoping auth method. Test all the operations with auth method set to ``token`` as a new, unified way for scoping all the tokens. """ AUTH_METHOD = 'token' def auth_plugin_config_override(self): methods = ['saml2', 'token'] super(FederatedTokenTests, self).auth_plugin_config_override(methods) def test_full_workflow(self): """Test 'standard' workflow for granting access tokens. * Issue unscoped token * List available projects based on groups * Scope token to one of available projects """ r = self._issue_unscoped_token() token_resp = render_token.render_token_response_from_model(r)['token'] # NOTE(lbragstad): Ensure only 'saml2' is in the method list. self.assertListEqual(['saml2'], r.methods) self.assertValidMappedUser(token_resp) employee_unscoped_token_id = r.id r = self.get('/auth/projects', token=employee_unscoped_token_id) projects = r.result['projects'] random_project = random.randint(0, len(projects) - 1) project = projects[random_project] v3_scope_request = self._scope_request(employee_unscoped_token_id, 'project', project['id']) r = self.v3_create_token(v3_scope_request) token_resp = r.result['token'] self.assertIn('token', token_resp['methods']) self.assertIn('saml2', token_resp['methods']) self._check_project_scoped_token_attributes(token_resp, project['id']) class FederatedUserTests(test_v3.RestfulTestCase, FederatedSetupMixin): """Test for federated users. Tests new shadow users functionality """ def auth_plugin_config_override(self): methods = ['saml2', 'token'] super(FederatedUserTests, self).auth_plugin_config_override(methods) def load_fixtures(self, fixtures): super(FederatedUserTests, self).load_fixtures(fixtures) self.load_federation_sample_data() def test_user_id_persistense(self): """Ensure user_id is persistend for multiple federated authn calls.""" r = self._issue_unscoped_token() user_id = r.user_id self.assertNotEmpty(PROVIDERS.identity_api.get_user(user_id)) r = self._issue_unscoped_token() user_id2 = r.user_id self.assertNotEmpty(PROVIDERS.identity_api.get_user(user_id2)) self.assertEqual(user_id, user_id2) def test_user_role_assignment(self): # create project and role project_ref = unit.new_project_ref( domain_id=CONF.identity.default_domain_id) PROVIDERS.resource_api.create_project(project_ref['id'], project_ref) role_ref = unit.new_role_ref() PROVIDERS.role_api.create_role(role_ref['id'], role_ref) # authenticate via saml get back a user id user_id, unscoped_token = self._authenticate_via_saml() # exchange an unscoped token for a scoped token; resulting in # unauthorized because the user doesn't have any role assignments v3_scope_request = self._scope_request(unscoped_token, 'project', project_ref['id']) r = self.v3_create_token(v3_scope_request, expected_status=http.client.UNAUTHORIZED) # assign project role to federated user PROVIDERS.assignment_api.add_role_to_user_and_project( user_id, project_ref['id'], role_ref['id']) # exchange an unscoped token for a scoped token r = self.v3_create_token(v3_scope_request, expected_status=http.client.CREATED) scoped_token = r.headers['X-Subject-Token'] # ensure user can access resource based on role assignment path = '/projects/%(project_id)s' % {'project_id': project_ref['id']} r = self.v3_request(path=path, method='GET', expected_status=http.client.OK, token=scoped_token) self.assertValidProjectResponse(r, project_ref) # create a 2nd project project_ref2 = unit.new_project_ref( domain_id=CONF.identity.default_domain_id) PROVIDERS.resource_api.create_project(project_ref2['id'], project_ref2) # ensure the user cannot access the 2nd resource (forbidden) path = '/projects/%(project_id)s' % {'project_id': project_ref2['id']} r = self.v3_request(path=path, method='GET', expected_status=http.client.FORBIDDEN, token=scoped_token) def test_domain_scoped_user_role_assignment(self): # create domain and role domain_ref = unit.new_domain_ref() PROVIDERS.resource_api.create_domain(domain_ref['id'], domain_ref) role_ref = unit.new_role_ref() PROVIDERS.role_api.create_role(role_ref['id'], role_ref) # authenticate via saml get back a user id user_id, unscoped_token = self._authenticate_via_saml() # exchange an unscoped token for a scoped token; resulting in # unauthorized because the user doesn't have any role assignments v3_scope_request = self._scope_request(unscoped_token, 'domain', domain_ref['id']) r = self.v3_create_token(v3_scope_request, expected_status=http.client.UNAUTHORIZED) # assign domain role to user PROVIDERS.assignment_api.create_grant( role_ref['id'], user_id=user_id, domain_id=domain_ref['id'] ) # exchange an unscoped token for domain scoped token and test r = self.v3_create_token(v3_scope_request, expected_status=http.client.CREATED) self.assertIsNotNone(r.headers.get('X-Subject-Token')) token_resp = r.result['token'] self.assertIn('domain', token_resp) def test_auth_projects_matches_federation_projects(self): # create project and role project_ref = unit.new_project_ref( domain_id=CONF.identity.default_domain_id) PROVIDERS.resource_api.create_project(project_ref['id'], project_ref) role_ref = unit.new_role_ref() PROVIDERS.role_api.create_role(role_ref['id'], role_ref) # authenticate via saml get back a user id user_id, unscoped_token = self._authenticate_via_saml() # assign project role to federated user PROVIDERS.assignment_api.add_role_to_user_and_project( user_id, project_ref['id'], role_ref['id']) # get auth projects r = self.get('/auth/projects', token=unscoped_token) auth_projects = r.result['projects'] # get federation projects r = self.get('/OS-FEDERATION/projects', token=unscoped_token) fed_projects = r.result['projects'] # compare self.assertCountEqual(auth_projects, fed_projects) def test_auth_projects_matches_federation_projects_with_group_assign(self): # create project, role, group domain_id = CONF.identity.default_domain_id project_ref = unit.new_project_ref(domain_id=domain_id) PROVIDERS.resource_api.create_project(project_ref['id'], project_ref) role_ref = unit.new_role_ref() PROVIDERS.role_api.create_role(role_ref['id'], role_ref) group_ref = unit.new_group_ref(domain_id=domain_id) group_ref = PROVIDERS.identity_api.create_group(group_ref) # authenticate via saml get back a user id user_id, unscoped_token = self._authenticate_via_saml() # assign role to group at project PROVIDERS.assignment_api.create_grant( role_ref['id'], group_id=group_ref['id'], project_id=project_ref['id'], domain_id=domain_id ) # add user to group PROVIDERS.identity_api.add_user_to_group( user_id=user_id,
{'commit': 'diff --git a/keystone/tests/unit/test_auth.py b/keystone/tests/unit/test_auth.py\nindex 6dd52c8a9..a0353c843 100644\n--- a/keystone/tests/unit/test_auth.py\n+++ b/keystone/tests/unit/test_auth.py\n@@ -682,7 +682,7 @@ class AuthWithPasswordCredentials(AuthTest):\n             {}, body_dict)\n \n     def test_authenticate_blank_password_credentials(self):\n-        """Sending empty dict as passwordCredentials raises a 400 error."""\n+        """Sending empty dict as passwordCredentials raises 400 Bad Requset."""\n         body_dict = {\'passwordCredentials\': {}, \'tenantName\': \'demo\'}\n         self.assertRaises(exception.ValidationError,\n                           self.controller.authenticate,\ndiff --git a/keystone/tests/unit/test_v2_keystoneclient.py b/keystone/tests/unit/test_v2_keystoneclient.py\nindex 50b7b8c9c..8367efdcb 100644\n--- a/keystone/tests/unit/test_v2_keystoneclient.py\n+++ b/keystone/tests/unit/test_v2_keystoneclient.py\n@@ -1083,7 +1083,7 @@ class ClientDrivenTestCase(unit.TestCase):\n         endpoint_internalurl = uuid.uuid4().hex\n         endpoint_adminurl = uuid.uuid4().hex\n \n-        # a non-existent service ID should trigger a 400\n+        # a non-existent service ID should trigger a 400 Bad Request\n         self.assertRaises(client_exceptions.BadRequest,\n                           client.endpoints.create,\n                           region=endpoint_region,\ndiff --git a/keystone/tests/unit/test_v3_assignment.py b/keystone/tests/unit/test_v3_assignment.py\nindex 71b6b89f2..541d364b6 100644\n--- a/keystone/tests/unit/test_v3_assignment.py\n+++ b/keystone/tests/unit/test_v3_assignment.py\n@@ -1903,11 +1903,11 @@ class RoleAssignmentBaseTestCase(test_v3.RestfulTestCase,\n class RoleAssignmentFailureTestCase(RoleAssignmentBaseTestCase):\n     """Class for testing invalid query params on /v3/role_assignments API.\n \n-    Querying domain and project, or user and group results in a HTTP 400, since\n-    a role assignment must contain only a single pair of (actor, target). In\n-    addition, since filtering on role assignments applies only to the final\n-    result, effective mode cannot be combined with i) group or ii) domain and\n-    inherited, because it would always result in an empty list.\n+    Querying domain and project, or user and group results in a HTTP 400 Bad\n+    Request, since a role assignment must contain only a single pair of (actor,\n+    target). In addition, since filtering on role assignments applies only to\n+    the final result, effective mode cannot be combined with i) group or ii)\n+    domain and inherited, because it would always result in an empty list.\n \n     """\n \ndiff --git a/keystone/tests/unit/test_v3_auth.py b/keystone/tests/unit/test_v3_auth.py\nindex 0adba23af..81998cd9f 100644\n--- a/keystone/tests/unit/test_v3_auth.py\n+++ b/keystone/tests/unit/test_v3_auth.py\n@@ -3183,7 +3183,8 @@ class TestTrustAuth(test_v3.RestfulTestCase):\n         self.trustee_user_id = self.trustee_user[\'id\']\n \n     def test_create_trust_bad_request(self):\n-        # The server returns a 403 Forbidden rather than a 400, see bug 1133435\n+        # The server returns a 403 Forbidden rather than a 400 Bad Request, see\n+        # bug 1133435\n         self.post(\'/OS-TRUST/trusts\', body={\'trust\': {}},\n                   expected_status=http_client.FORBIDDEN)\n \ndiff --git a/keystone/tests/unit/test_v3_federation.py b/keystone/tests/unit/test_v3_federation.py\nindex 2de4b9970..011ba7f62 100644\n--- a/keystone/tests/unit/test_v3_federation.py\n+++ b/keystone/tests/unit/test_v3_federation.py\n@@ -1469,8 +1469,8 @@ class MappingCRUDTests(FederationTests):\n     def test_create_mapping_with_blacklist_and_whitelist(self):\n         """Test for adding whitelist and blacklist in the rule\n \n-        Server should respond with HTTP 400 error upon discovering both\n-        ``whitelist`` and ``blacklist`` keywords in the same rule.\n+        Server should respond with HTTP 400 Bad Request error upon discovering\n+        both ``whitelist`` and ``blacklist`` keywords in the same rule.\n \n         """\n         url = self.MAPPING_URL + uuid.uuid4().hex\n@@ -2796,7 +2796,7 @@ class SAMLGenerationTests(FederationTests):\n     def test_invalid_scope_body(self):\n         """Test that missing the scope in request body raises an exception.\n \n-        Raises exception.SchemaValidationError() - error code 400\n+        Raises exception.SchemaValidationError() - error 400 Bad Request\n \n         """\n \n@@ -2811,7 +2811,7 @@ class SAMLGenerationTests(FederationTests):\n     def test_invalid_token_body(self):\n         """Test that missing the token in request body raises an exception.\n \n-        Raises exception.SchemaValidationError() - error code 400\n+        Raises exception.SchemaValidationError() - error 400 Bad Request\n \n         """\n \n', 'commit_subject': 'Merge "Refactor: change 400 status codes in test names"', 'repo_name': 'openstack/keystone', 'committer': {'name': 'Gerrit Code Review', 'email': '61e62b213a1a56f7695845df4fc372a10cb0a73e@openstack.org', 'time_sec': '1444626427', 'tz_offset': '0', 'date': {'seconds': '1444626427'}}, 'commit_tree': '7b13e57687d374f361a97441a672b5c1922dc5f9', 'commit_message': 'Merge "Refactor: change 400 status codes in test names"', 'language_name': 'Python', 'license': 'apache-2.0', 'hunks': '@@ -1469,8 +1469,8 @@ class MappingCRUDTests(FederationTests):\n     def test_create_mapping_with_blacklist_and_whitelist(self):\n         """Test for adding whitelist and blacklist in the rule\n \n-        Server should respond with HTTP 400 error upon discovering both\n-        ``whitelist`` and ``blacklist`` keywords in the same rule.\n+        Server should respond with HTTP 400 Bad Request error upon discovering\n+        both ``whitelist`` and ``blacklist`` keywords in the same rule.\n \n         """\n         url = self.MAPPING_URL + uuid.uuid4().hex\n@@ -2796,7 +2796,7 @@ class SAMLGenerationTests(FederationTests):\n     def test_invalid_scope_body(self):\n         """Test that missing the scope in request body raises an exception.\n \n-        Raises exception.SchemaValidationError() - error code 400\n+        Raises exception.SchemaValidationError() - error 400 Bad Request\n \n         """\n \n@@ -2811,7 +2811,7 @@ class SAMLGenerationTests(FederationTests):\n     def test_invalid_token_body(self):\n         """Test that missing the token in request body raises an exception.\n \n-        Raises exception.SchemaValidationError() - error code 400\n+        Raises exception.SchemaValidationError() - error 400 Bad Request\n \n         """\n \n', 'addition_count': 4, 'deletion_count': 4, 'src_file': 'a/keystone/tests/unit/test_v3_federation.py', 'tgt_file': 'b/keystone/tests/unit/test_v3_federation.py', 'file_extension': '.py', 'before_file': '# Licensed under the Apache License, Version 2.0 (the "License"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\nimport copy\nimport os\nimport random\nimport re\nimport subprocess\nfrom testtools import matchers\nfrom unittest import mock\nimport uuid\n\nimport fixtures\nimport flask\nimport http.client\nfrom lxml import etree\nfrom oslo_serialization import jsonutils\nfrom oslo_utils import importutils\nimport saml2\nfrom saml2 import saml\nfrom saml2 import sigver\nimport urllib\nxmldsig = importutils.try_import("saml2.xmldsig")\nif not xmldsig:\n    xmldsig = importutils.try_import("xmldsig")\n\nfrom keystone.api._shared import authentication\nfrom keystone.api import auth as auth_api\nfrom keystone.common import driver_hints\nfrom keystone.common import provider_api\nfrom keystone.common import render_token\nimport keystone.conf\nfrom keystone import exception\nfrom keystone.federation import idp as keystone_idp\nfrom keystone.models import token_model\nfrom keystone import notifications\nfrom keystone.tests import unit\nfrom keystone.tests.unit import core\nfrom keystone.tests.unit import federation_fixtures\nfrom keystone.tests.unit import ksfixtures\nfrom keystone.tests.unit import mapping_fixtures\nfrom keystone.tests.unit import test_v3\n\n\nCONF = keystone.conf.CONF\nPROVIDERS = provider_api.ProviderAPIs\nROOTDIR = os.path.dirname(os.path.abspath(__file__))\nXMLDIR = os.path.join(ROOTDIR, \'saml2/\')\n\n\ndef dummy_validator(*args, **kwargs):\n    pass\n\n\nclass FederatedSetupMixin(object):\n\n    ACTION = \'authenticate\'\n    IDP = \'ORG_IDP\'\n    PROTOCOL = \'saml2\'\n    AUTH_METHOD = \'saml2\'\n    USER = \'user@ORGANIZATION\'\n    ASSERTION_PREFIX = \'PREFIX_\'\n    IDP_WITH_REMOTE = \'ORG_IDP_REMOTE\'\n    REMOTE_IDS = [\'entityID_IDP1\', \'entityID_IDP2\']\n    REMOTE_ID_ATTR = uuid.uuid4().hex\n\n    UNSCOPED_V3_SAML2_REQ = {\n        "identity": {\n            "methods": [AUTH_METHOD],\n            AUTH_METHOD: {\n                "identity_provider": IDP,\n                "protocol": PROTOCOL\n            }\n        }\n    }\n\n    def _check_domains_are_valid(self, token):\n        domain = PROVIDERS.resource_api.get_domain(self.idp[\'domain_id\'])\n        self.assertEqual(domain[\'id\'], token[\'user\'][\'domain\'][\'id\'])\n        self.assertEqual(domain[\'name\'], token[\'user\'][\'domain\'][\'name\'])\n\n    def _project(self, project):\n        return (project[\'id\'], project[\'name\'])\n\n    def _roles(self, roles):\n        return set([(r[\'id\'], r[\'name\']) for r in roles])\n\n    def _check_projects_and_roles(self, token, roles, projects):\n        """Check whether the projects and the roles match."""\n        token_roles = token.get(\'roles\')\n        if token_roles is None:\n            raise AssertionError(\'Roles not found in the token\')\n        token_roles = self._roles(token_roles)\n        roles_ref = self._roles(roles)\n        self.assertEqual(token_roles, roles_ref)\n\n        token_projects = token.get(\'project\')\n        if token_projects is None:\n            raise AssertionError(\'Projects not found in the token\')\n        token_projects = self._project(token_projects)\n        projects_ref = self._project(projects)\n        self.assertEqual(token_projects, projects_ref)\n\n    def _check_scoped_token_attributes(self, token):\n\n        for obj in (\'user\', \'catalog\', \'expires_at\', \'issued_at\',\n                    \'methods\', \'roles\'):\n            self.assertIn(obj, token)\n\n        os_federation = token[\'user\'][\'OS-FEDERATION\']\n\n        self.assertIn(\'groups\', os_federation)\n        self.assertIn(\'identity_provider\', os_federation)\n        self.assertIn(\'protocol\', os_federation)\n        self.assertThat(os_federation, matchers.HasLength(3))\n\n        self.assertEqual(self.IDP, os_federation[\'identity_provider\'][\'id\'])\n        self.assertEqual(self.PROTOCOL, os_federation[\'protocol\'][\'id\'])\n\n    def _check_project_scoped_token_attributes(self, token, project_id):\n        self.assertEqual(project_id, token[\'project\'][\'id\'])\n        self._check_scoped_token_attributes(token)\n\n    def _check_domain_scoped_token_attributes(self, token, domain_id):\n        self.assertEqual(domain_id, token[\'domain\'][\'id\'])\n        self._check_scoped_token_attributes(token)\n\n    def assertValidMappedUser(self, token):\n        """Check if user object meets all the criteria."""\n        user = token[\'user\']\n        self.assertIn(\'id\', user)\n        self.assertIn(\'name\', user)\n        self.assertIn(\'domain\', user)\n\n        self.assertIn(\'groups\', user[\'OS-FEDERATION\'])\n        self.assertIn(\'identity_provider\', user[\'OS-FEDERATION\'])\n        self.assertIn(\'protocol\', user[\'OS-FEDERATION\'])\n\n        # Make sure user_name is url safe\n        self.assertEqual(urllib.parse.quote(user[\'name\']), user[\'name\'])\n\n    def _issue_unscoped_token(self,\n                              idp=None,\n                              assertion=\'EMPLOYEE_ASSERTION\',\n                              environment=None):\n        environment = environment or {}\n        environment.update(getattr(mapping_fixtures, assertion))\n        with self.make_request(environ=environment):\n            if idp is None:\n                idp = self.IDP\n            r = authentication.federated_authenticate_for_token(\n                protocol_id=self.PROTOCOL, identity_provider=idp)\n        return r\n\n    def idp_ref(self, id=None):\n        idp = {\n            \'id\': id or uuid.uuid4().hex,\n            \'enabled\': True,\n            \'description\': uuid.uuid4().hex\n        }\n        return idp\n\n    def proto_ref(self, mapping_id=None):\n        proto = {\n            \'id\': uuid.uuid4().hex,\n            \'mapping_id\': mapping_id or uuid.uuid4().hex\n        }\n        return proto\n\n    def mapping_ref(self, rules=None):\n        return {\n            \'id\': uuid.uuid4().hex,\n            \'rules\': rules or self.rules[\'rules\']\n        }\n\n    def _scope_request(self, unscoped_token_id, scope, scope_id):\n        return {\n            \'auth\': {\n                \'identity\': {\n                    \'methods\': [\n                        \'token\'\n                    ],\n                    \'token\': {\n                        \'id\': unscoped_token_id\n                    }\n                },\n                \'scope\': {\n                    scope: {\n                        \'id\': scope_id\n                    }\n                }\n            }\n        }\n\n    def _inject_assertion(self, variant):\n        assertion = getattr(mapping_fixtures, variant)\n        flask.request.environ.update(assertion)\n\n    def load_federation_sample_data(self):\n        """Inject additional data."""\n        # Create and add domains\n        self.domainA = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(\n            self.domainA[\'id\'], self.domainA\n        )\n\n        self.domainB = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(\n            self.domainB[\'id\'], self.domainB\n        )\n\n        self.domainC = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(\n            self.domainC[\'id\'], self.domainC\n        )\n\n        self.domainD = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(\n            self.domainD[\'id\'], self.domainD\n        )\n\n        # Create and add projects\n        self.proj_employees = unit.new_project_ref(\n            domain_id=self.domainA[\'id\'])\n        PROVIDERS.resource_api.create_project(\n            self.proj_employees[\'id\'], self.proj_employees\n        )\n        self.proj_customers = unit.new_project_ref(\n            domain_id=self.domainA[\'id\'])\n        PROVIDERS.resource_api.create_project(\n            self.proj_customers[\'id\'], self.proj_customers\n        )\n\n        self.project_all = unit.new_project_ref(\n            domain_id=self.domainA[\'id\'])\n        PROVIDERS.resource_api.create_project(\n            self.project_all[\'id\'], self.project_all\n        )\n\n        self.project_inherited = unit.new_project_ref(\n            domain_id=self.domainD[\'id\'])\n        PROVIDERS.resource_api.create_project(\n            self.project_inherited[\'id\'], self.project_inherited\n        )\n\n        # Create and add groups\n        self.group_employees = unit.new_group_ref(domain_id=self.domainA[\'id\'])\n        self.group_employees = (\n            PROVIDERS.identity_api.create_group(self.group_employees))\n\n        self.group_customers = unit.new_group_ref(domain_id=self.domainA[\'id\'])\n        self.group_customers = (\n            PROVIDERS.identity_api.create_group(self.group_customers))\n\n        self.group_admins = unit.new_group_ref(domain_id=self.domainA[\'id\'])\n        self.group_admins = PROVIDERS.identity_api.create_group(\n            self.group_admins\n        )\n\n        # Create and add roles\n        self.role_employee = unit.new_role_ref()\n        PROVIDERS.role_api.create_role(\n            self.role_employee[\'id\'], self.role_employee\n        )\n        self.role_customer = unit.new_role_ref()\n        PROVIDERS.role_api.create_role(\n            self.role_customer[\'id\'], self.role_customer\n        )\n\n        self.role_admin = unit.new_role_ref()\n        PROVIDERS.role_api.create_role(self.role_admin[\'id\'], self.role_admin)\n\n        # Employees can access\n        # * proj_employees\n        # * project_all\n        PROVIDERS.assignment_api.create_grant(\n            self.role_employee[\'id\'], group_id=self.group_employees[\'id\'],\n            project_id=self.proj_employees[\'id\']\n        )\n        PROVIDERS.assignment_api.create_grant(\n            self.role_employee[\'id\'], group_id=self.group_employees[\'id\'],\n            project_id=self.project_all[\'id\']\n        )\n        # Customers can access\n        # * proj_customers\n        PROVIDERS.assignment_api.create_grant(\n            self.role_customer[\'id\'], group_id=self.group_customers[\'id\'],\n            project_id=self.proj_customers[\'id\']\n        )\n\n        # Admins can access:\n        # * proj_customers\n        # * proj_employees\n        # * project_all\n        PROVIDERS.assignment_api.create_grant(\n            self.role_admin[\'id\'], group_id=self.group_admins[\'id\'],\n            project_id=self.proj_customers[\'id\']\n        )\n        PROVIDERS.assignment_api.create_grant(\n            self.role_admin[\'id\'], group_id=self.group_admins[\'id\'],\n            project_id=self.proj_employees[\'id\']\n        )\n        PROVIDERS.assignment_api.create_grant(\n            self.role_admin[\'id\'], group_id=self.group_admins[\'id\'],\n            project_id=self.project_all[\'id\']\n        )\n\n        # Customers can access:\n        # * domain A\n        PROVIDERS.assignment_api.create_grant(\n            self.role_customer[\'id\'], group_id=self.group_customers[\'id\'],\n            domain_id=self.domainA[\'id\']\n        )\n\n        # Customers can access projects via inheritance:\n        # * domain D\n        PROVIDERS.assignment_api.create_grant(\n            self.role_customer[\'id\'], group_id=self.group_customers[\'id\'],\n            domain_id=self.domainD[\'id\'], inherited_to_projects=True\n        )\n\n        # Employees can access:\n        # * domain A\n        # * domain B\n\n        PROVIDERS.assignment_api.create_grant(\n            self.role_employee[\'id\'], group_id=self.group_employees[\'id\'],\n            domain_id=self.domainA[\'id\']\n        )\n        PROVIDERS.assignment_api.create_grant(\n            self.role_employee[\'id\'], group_id=self.group_employees[\'id\'],\n            domain_id=self.domainB[\'id\']\n        )\n\n        # Admins can access:\n        # * domain A\n        # * domain B\n        # * domain C\n        PROVIDERS.assignment_api.create_grant(\n            self.role_admin[\'id\'], group_id=self.group_admins[\'id\'],\n            domain_id=self.domainA[\'id\']\n        )\n        PROVIDERS.assignment_api.create_grant(\n            self.role_admin[\'id\'], group_id=self.group_admins[\'id\'],\n            domain_id=self.domainB[\'id\']\n        )\n\n        PROVIDERS.assignment_api.create_grant(\n            self.role_admin[\'id\'], group_id=self.group_admins[\'id\'],\n            domain_id=self.domainC[\'id\']\n        )\n        self.rules = {\n            \'rules\': [\n                {\n                    \'local\': [\n                        {\n                            \'group\': {\n                                \'id\': self.group_employees[\'id\']\n                            }\n                        },\n                        {\n                            \'user\': {\n                                \'name\': \'{0}\',\n                                \'id\': \'{1}\'\n                            }\n                        }\n                    ],\n                    \'remote\': [\n                        {\n                            \'type\': \'UserName\'\n                        },\n                        {\n                            \'type\': \'Email\',\n                        },\n                        {\n                            \'type\': \'orgPersonType\',\n                            \'any_one_of\': [\n                                \'Employee\'\n                            ]\n                        }\n                    ]\n                },\n                {\n                    \'local\': [\n                        {\n                            \'group\': {\n                                \'id\': self.group_employees[\'id\']\n                            }\n                        },\n                        {\n                            \'user\': {\n                                \'name\': \'{0}\',\n                                \'id\': \'{1}\'\n                            }\n                        }\n                    ],\n                    \'remote\': [\n                        {\n                            \'type\': self.ASSERTION_PREFIX + \'UserName\'\n                        },\n                        {\n                            \'type\': self.ASSERTION_PREFIX + \'Email\',\n                        },\n                        {\n                            \'type\': self.ASSERTION_PREFIX + \'orgPersonType\',\n                            \'any_one_of\': [\n                                \'SuperEmployee\'\n                            ]\n                        }\n                    ]\n                },\n                {\n                    \'local\': [\n                        {\n                            \'group\': {\n                                \'id\': self.group_customers[\'id\']\n                            }\n                        },\n                        {\n                            \'user\': {\n                                \'name\': \'{0}\',\n                                \'id\': \'{1}\'\n                            }\n                        }\n                    ],\n                    \'remote\': [\n                        {\n                            \'type\': \'UserName\'\n                        },\n                        {\n                            \'type\': \'Email\'\n                        },\n                        {\n                            \'type\': \'orgPersonType\',\n                            \'any_one_of\': [\n                                \'Customer\'\n                            ]\n                        }\n                    ]\n                },\n                {\n                    \'local\': [\n                        {\n                            \'group\': {\n                                \'id\': self.group_admins[\'id\']\n                            }\n                        },\n                        {\n                            \'group\': {\n                                \'id\': self.group_employees[\'id\']\n                            }\n                        },\n                        {\n                            \'group\': {\n                                \'id\': self.group_customers[\'id\']\n                            }\n                        },\n\n                        {\n                            \'user\': {\n                                \'name\': \'{0}\',\n                                \'id\': \'{1}\'\n                            }\n                        }\n                    ],\n                    \'remote\': [\n                        {\n                            \'type\': \'UserName\'\n                        },\n                        {\n                            \'type\': \'Email\'\n                        },\n                        {\n                            \'type\': \'orgPersonType\',\n                            \'any_one_of\': [\n                                \'Admin\',\n                                \'Chief\'\n                            ]\n                        }\n                    ]\n                },\n                {\n                    \'local\': [\n                        {\n                            \'group\': {\n                                \'id\': uuid.uuid4().hex\n                            }\n                        },\n                        {\n                            \'group\': {\n                                \'id\': self.group_customers[\'id\']\n                            }\n                        },\n                        {\n                            \'user\': {\n                                \'name\': \'{0}\',\n                                \'id\': \'{1}\'\n                            }\n                        }\n                    ],\n                    \'remote\': [\n                        {\n                            \'type\': \'UserName\',\n                        },\n                        {\n                            \'type\': \'Email\',\n                        },\n                        {\n                            \'type\': \'FirstName\',\n                            \'any_one_of\': [\n                                \'Jill\'\n                            ]\n                        },\n                        {\n                            \'type\': \'LastName\',\n                            \'any_one_of\': [\n                                \'Smith\'\n                            ]\n                        }\n                    ]\n                },\n                {\n                    \'local\': [\n                        {\n                            \'group\': {\n                                \'id\': \'this_group_no_longer_exists\'\n                            }\n                        },\n                        {\n                            \'user\': {\n                                \'name\': \'{0}\',\n                                \'id\': \'{1}\'\n                            }\n                        }\n                    ],\n                    \'remote\': [\n                        {\n                            \'type\': \'UserName\',\n                        },\n                        {\n                            \'type\': \'Email\',\n                        },\n                        {\n                            \'type\': \'Email\',\n                            \'any_one_of\': [\n                                \'testacct@example.com\'\n                            ]\n                        },\n                        {\n                            \'type\': \'orgPersonType\',\n                            \'any_one_of\': [\n                                \'Tester\'\n                            ]\n                        }\n                    ]\n                },\n                # rules with local group names\n                {\n                    "local": [\n                        {\n                            \'user\': {\n                                \'name\': \'{0}\',\n                                \'id\': \'{1}\'\n                            }\n                        },\n                        {\n                            "group": {\n                                "name": self.group_customers[\'name\'],\n                                "domain": {\n                                    "name": self.domainA[\'name\']\n                                }\n                            }\n                        }\n                    ],\n                    "remote": [\n                        {\n                            \'type\': \'UserName\',\n                        },\n                        {\n                            \'type\': \'Email\',\n                        },\n                        {\n                            "type": "orgPersonType",\n                            "any_one_of": [\n                                "CEO",\n                                "CTO"\n                            ],\n                        }\n                    ]\n                },\n                {\n                    "local": [\n                        {\n                            \'user\': {\n                                \'name\': \'{0}\',\n                                \'id\': \'{1}\'\n                            }\n                        },\n                        {\n                            "group": {\n                                "name": self.group_admins[\'name\'],\n                                "domain": {\n                                    "id": self.domainA[\'id\']\n                                }\n                            }\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "UserName",\n                        },\n                        {\n                            "type": "Email",\n                        },\n                        {\n                            "type": "orgPersonType",\n                            "any_one_of": [\n                                "Managers"\n                            ]\n                        }\n                    ]\n                },\n                {\n                    "local": [\n                        {\n                            "user": {\n                                "name": "{0}",\n                                "id": "{1}"\n                            }\n                        },\n                        {\n                            "group": {\n                                "name": "NON_EXISTING",\n                                "domain": {\n                                    "id": self.domainA[\'id\']\n                                }\n                            }\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "UserName",\n                        },\n                        {\n                            "type": "Email",\n                        },\n                        {\n                            "type": "UserName",\n                            "any_one_of": [\n                                "IamTester"\n                            ]\n                        }\n                    ]\n                },\n                {\n                    "local": [\n                        {\n                            "user": {\n                                "type": "local",\n                                "name": self.user[\'name\'],\n                                "domain": {\n                                    "id": self.user[\'domain_id\']\n                                }\n                            }\n                        },\n                        {\n                            "group": {\n                                "id": self.group_customers[\'id\']\n                            }\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "UserType",\n                            "any_one_of": [\n                                "random"\n                            ]\n                        }\n                    ]\n                },\n                {\n                    "local": [\n                        {\n                            "user": {\n                                "type": "local",\n                                "name": self.user[\'name\'],\n                                "domain": {\n                                    "id": uuid.uuid4().hex\n                                }\n                            }\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "Position",\n                            "any_one_of": [\n                                "DirectorGeneral"\n                            ]\n                        }\n                    ]\n                },\n                # rules for users with no groups\n                {\n                    "local": [\n                        {\n                            \'user\': {\n                                \'name\': \'{0}\',\n                                \'id\': \'{1}\'\n                            }\n                        }\n                    ],\n                    "remote": [\n                        {\n                            \'type\': \'UserName\',\n                        },\n                        {\n                            \'type\': \'Email\',\n                        },\n                        {\n                            \'type\': \'orgPersonType\',\n                            \'any_one_of\': [\n                                \'NoGroupsOrg\'\n                            ]\n                        }\n                    ]\n                }\n            ]\n        }\n\n        # Add unused IdP first so it is indexed first (#1838592)\n        self.dummy_idp = self.idp_ref()\n        PROVIDERS.federation_api.create_idp(\n            self.dummy_idp[\'id\'], self.dummy_idp\n        )\n        # Add IDP\n        self.idp = self.idp_ref(id=self.IDP)\n        PROVIDERS.federation_api.create_idp(\n            self.idp[\'id\'], self.idp\n        )\n        # Add IDP with remote\n        self.idp_with_remote = self.idp_ref(id=self.IDP_WITH_REMOTE)\n        self.idp_with_remote[\'remote_ids\'] = self.REMOTE_IDS\n        PROVIDERS.federation_api.create_idp(\n            self.idp_with_remote[\'id\'], self.idp_with_remote\n        )\n        # Add a mapping\n        self.mapping = self.mapping_ref()\n        PROVIDERS.federation_api.create_mapping(\n            self.mapping[\'id\'], self.mapping\n        )\n        # Add protocols\n        self.proto_saml = self.proto_ref(mapping_id=self.mapping[\'id\'])\n        self.proto_saml[\'id\'] = self.PROTOCOL\n        PROVIDERS.federation_api.create_protocol(\n            self.idp[\'id\'], self.proto_saml[\'id\'], self.proto_saml\n        )\n        # Add protocols IDP with remote\n        PROVIDERS.federation_api.create_protocol(\n            self.idp_with_remote[\'id\'], self.proto_saml[\'id\'], self.proto_saml\n        )\n        # Add unused protocol to go with unused IdP (#1838592)\n        self.proto_dummy = self.proto_ref(mapping_id=self.mapping[\'id\'])\n        PROVIDERS.federation_api.create_protocol(\n            self.dummy_idp[\'id\'], self.proto_dummy[\'id\'], self.proto_dummy\n        )\n\n        with self.make_request():\n            self.tokens = {}\n            VARIANTS = (\'EMPLOYEE_ASSERTION\', \'CUSTOMER_ASSERTION\',\n                        \'ADMIN_ASSERTION\')\n            for variant in VARIANTS:\n                self._inject_assertion(variant)\n                r = authentication.authenticate_for_token(\n                    self.UNSCOPED_V3_SAML2_REQ)\n                self.tokens[variant] = r.id\n\n            self.TOKEN_SCOPE_PROJECT_FROM_NONEXISTENT_TOKEN = (\n                self._scope_request(\n                    uuid.uuid4().hex, \'project\', self.proj_customers[\'id\']))\n\n            self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_EMPLOYEE = (\n                self._scope_request(\n                    self.tokens[\'EMPLOYEE_ASSERTION\'], \'project\',\n                    self.proj_employees[\'id\']))\n\n            self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN = self._scope_request(\n                self.tokens[\'ADMIN_ASSERTION\'], \'project\',\n                self.proj_employees[\'id\'])\n\n            self.TOKEN_SCOPE_PROJECT_CUSTOMER_FROM_ADMIN = self._scope_request(\n                self.tokens[\'ADMIN_ASSERTION\'], \'project\',\n                self.proj_customers[\'id\'])\n\n            self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_CUSTOMER = (\n                self._scope_request(\n                    self.tokens[\'CUSTOMER_ASSERTION\'], \'project\',\n                    self.proj_employees[\'id\']))\n\n            self.TOKEN_SCOPE_PROJECT_INHERITED_FROM_CUSTOMER = (\n                self._scope_request(\n                    self.tokens[\'CUSTOMER_ASSERTION\'], \'project\',\n                    self.project_inherited[\'id\']))\n\n            self.TOKEN_SCOPE_DOMAIN_A_FROM_CUSTOMER = self._scope_request(\n                self.tokens[\'CUSTOMER_ASSERTION\'], \'domain\',\n                self.domainA[\'id\'])\n\n            self.TOKEN_SCOPE_DOMAIN_B_FROM_CUSTOMER = self._scope_request(\n                self.tokens[\'CUSTOMER_ASSERTION\'], \'domain\',\n                self.domainB[\'id\'])\n\n            self.TOKEN_SCOPE_DOMAIN_D_FROM_CUSTOMER = self._scope_request(\n                self.tokens[\'CUSTOMER_ASSERTION\'], \'domain\',\n                self.domainD[\'id\'])\n\n            self.TOKEN_SCOPE_DOMAIN_A_FROM_ADMIN = self._scope_request(\n                self.tokens[\'ADMIN_ASSERTION\'], \'domain\', self.domainA[\'id\'])\n\n            self.TOKEN_SCOPE_DOMAIN_B_FROM_ADMIN = self._scope_request(\n                self.tokens[\'ADMIN_ASSERTION\'], \'domain\', self.domainB[\'id\'])\n\n            self.TOKEN_SCOPE_DOMAIN_C_FROM_ADMIN = self._scope_request(\n                self.tokens[\'ADMIN_ASSERTION\'], \'domain\',\n                self.domainC[\'id\'])\n\n\nclass FederatedIdentityProviderTests(test_v3.RestfulTestCase):\n    """A test class for Identity Providers."""\n\n    idp_keys = [\'description\', \'enabled\']\n\n    default_body = {\'description\': None, \'enabled\': True}\n\n    def base_url(self, suffix=None):\n        if suffix is not None:\n            return \'/OS-FEDERATION/identity_providers/\' + str(suffix)\n        return \'/OS-FEDERATION/identity_providers\'\n\n    def _fetch_attribute_from_response(self, resp, parameter,\n                                       assert_is_not_none=True):\n        """Fetch single attribute from TestResponse object."""\n        result = resp.result.get(parameter)\n        if assert_is_not_none:\n            self.assertIsNotNone(result)\n        return result\n\n    def _create_and_decapsulate_response(self, body=None):\n        """Create IdP and fetch it\'s random id along with entity."""\n        default_resp = self._create_default_idp(body=body)\n        idp = self._fetch_attribute_from_response(default_resp,\n                                                  \'identity_provider\')\n        self.assertIsNotNone(idp)\n        idp_id = idp.get(\'id\')\n        return (idp_id, idp)\n\n    def _get_idp(self, idp_id):\n        """Fetch IdP entity based on its id."""\n        url = self.base_url(suffix=idp_id)\n        resp = self.get(url)\n        return resp\n\n    def _create_default_idp(self, body=None,\n                            expected_status=http.client.CREATED):\n        """Create default IdP."""\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        if body is None:\n            body = self._http_idp_input()\n        resp = self.put(url, body={\'identity_provider\': body},\n                        expected_status=expected_status)\n        return resp\n\n    def _http_idp_input(self):\n        """Create default input dictionary for IdP data."""\n        body = self.default_body.copy()\n        body[\'description\'] = uuid.uuid4().hex\n        return body\n\n    def _assign_protocol_to_idp(self, idp_id=None, proto=None, url=None,\n                                mapping_id=None, validate=True, **kwargs):\n        if url is None:\n            url = self.base_url(suffix=\'%(idp_id)s/protocols/%(protocol_id)s\')\n        if idp_id is None:\n            idp_id, _ = self._create_and_decapsulate_response()\n        if proto is None:\n            proto = uuid.uuid4().hex\n        if mapping_id is None:\n            mapping_id = uuid.uuid4().hex\n        self._create_mapping(mapping_id)\n        body = {\'mapping_id\': mapping_id}\n        url = url % {\'idp_id\': idp_id, \'protocol_id\': proto}\n        resp = self.put(url, body={\'protocol\': body}, **kwargs)\n        if validate:\n            self.assertValidResponse(resp, \'protocol\', dummy_validator,\n                                     keys_to_check=[\'id\', \'mapping_id\'],\n                                     ref={\'id\': proto,\n                                          \'mapping_id\': mapping_id})\n        return (resp, idp_id, proto)\n\n    def _get_protocol(self, idp_id, protocol_id):\n        url = \'%s/protocols/%s\' % (idp_id, protocol_id)\n        url = self.base_url(suffix=url)\n        r = self.get(url)\n        return r\n\n    def _create_mapping(self, mapping_id):\n        mapping = mapping_fixtures.MAPPING_EPHEMERAL_USER\n        mapping[\'id\'] = mapping_id\n        url = \'/OS-FEDERATION/mappings/%s\' % mapping_id\n        self.put(url,\n                 body={\'mapping\': mapping},\n                 expected_status=http.client.CREATED)\n\n    def assertIdpDomainCreated(self, idp_id, domain_id):\n        domain = PROVIDERS.resource_api.get_domain(domain_id)\n        self.assertEqual(domain_id, domain[\'name\'])\n        self.assertIn(idp_id, domain[\'description\'])\n\n    def test_create_idp_without_domain_id(self):\n        """Create the IdentityProvider entity associated to remote_ids."""\n        keys_to_check = list(self.idp_keys)\n        body = self.default_body.copy()\n        body[\'description\'] = uuid.uuid4().hex\n        resp = self._create_default_idp(body=body)\n        self.assertValidResponse(resp, \'identity_provider\', dummy_validator,\n                                 keys_to_check=keys_to_check,\n                                 ref=body)\n        attr = self._fetch_attribute_from_response(resp, \'identity_provider\')\n        self.assertIdpDomainCreated(attr[\'id\'], attr[\'domain_id\'])\n\n    def test_create_idp_with_domain_id(self):\n        keys_to_check = list(self.idp_keys)\n        keys_to_check.append(\'domain_id\')\n        body = self.default_body.copy()\n        body[\'description\'] = uuid.uuid4().hex\n        domain = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(domain[\'id\'], domain)\n        body[\'domain_id\'] = domain[\'id\']\n        resp = self._create_default_idp(body=body)\n        self.assertValidResponse(resp, \'identity_provider\', dummy_validator,\n                                 keys_to_check=keys_to_check,\n                                 ref=body)\n\n    def test_create_idp_domain_id_none(self):\n        keys_to_check = list(self.idp_keys)\n        body = self.default_body.copy()\n        body[\'description\'] = uuid.uuid4().hex\n        body[\'domain_id\'] = None\n        resp = self._create_default_idp(body=body)\n        self.assertValidResponse(resp, \'identity_provider\', dummy_validator,\n                                 keys_to_check=keys_to_check,\n                                 ref=body)\n        attr = self._fetch_attribute_from_response(resp, \'identity_provider\')\n        self.assertIdpDomainCreated(attr[\'id\'], attr[\'domain_id\'])\n\n    def test_conflicting_idp_cleans_up_auto_generated_domain(self):\n        # NOTE(lbragstad): Create an identity provider, save its ID, and count\n        # the number of domains.\n        resp = self._create_default_idp()\n        idp_id = resp.json_body[\'identity_provider\'][\'id\']\n        domains = PROVIDERS.resource_api.list_domains()\n        number_of_domains = len(domains)\n\n        # Create an identity provider with the same ID to intentionally cause a\n        # conflict, this is going to result in a domain getting created for the\n        # new identity provider. The domain for the new identity provider is\n        # going to be created before the conflict is raised from the database\n        # layer. This makes sure the domain is cleaned up after a Conflict is\n        # detected.\n        resp = self.put(\n            self.base_url(suffix=idp_id),\n            body={\'identity_provider\': self.default_body.copy()},\n            expected_status=http.client.CONFLICT\n        )\n        domains = PROVIDERS.resource_api.list_domains()\n        self.assertEqual(number_of_domains, len(domains))\n\n    def test_conflicting_idp_does_not_delete_existing_domain(self):\n        # Create a new domain\n        domain = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(domain[\'id\'], domain)\n\n        # Create an identity provider and specify the domain\n        body = self.default_body.copy()\n        body[\'description\'] = uuid.uuid4().hex\n        body[\'domain_id\'] = domain[\'id\']\n        resp = self._create_default_idp(body=body)\n        idp = resp.json_body[\'identity_provider\']\n        idp_id = idp[\'id\']\n        self.assertEqual(idp[\'domain_id\'], domain[\'id\'])\n\n        # Create an identity provider with the same domain and ID to ensure a\n        # Conflict is raised and then to verify the existing domain not deleted\n        # below\n        body = self.default_body.copy()\n        body[\'domain_id\'] = domain[\'id\']\n        resp = self.put(\n            self.base_url(suffix=idp_id),\n            body={\'identity_provider\': body},\n            expected_status=http.client.CONFLICT\n        )\n\n        # Make sure the domain specified in the second request was not deleted,\n        # since it wasn\'t auto-generated\n        self.assertIsNotNone(PROVIDERS.resource_api.get_domain(domain[\'id\']))\n\n    def test_create_multi_idp_to_one_domain(self):\n        # create domain and add domain_id to keys to check\n        domain = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(domain[\'id\'], domain)\n        keys_to_check = list(self.idp_keys)\n        keys_to_check.append(\'domain_id\')\n        # create idp with the domain_id\n        body = self.default_body.copy()\n        body[\'description\'] = uuid.uuid4().hex\n        body[\'domain_id\'] = domain[\'id\']\n        idp1 = self._create_default_idp(body=body)\n        self.assertValidResponse(idp1, \'identity_provider\', dummy_validator,\n                                 keys_to_check=keys_to_check,\n                                 ref=body)\n        # create a 2nd idp with the same domain_id\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        body = self.default_body.copy()\n        body[\'description\'] = uuid.uuid4().hex\n        body[\'domain_id\'] = domain[\'id\']\n        idp2 = self.put(url, body={\'identity_provider\': body},\n                        expected_status=http.client.CREATED)\n        self.assertValidResponse(idp2, \'identity_provider\', dummy_validator,\n                                 keys_to_check=keys_to_check,\n                                 ref=body)\n\n        self.assertEqual(idp1.result[\'identity_provider\'][\'domain_id\'],\n                         idp2.result[\'identity_provider\'][\'domain_id\'])\n\n    def test_cannot_update_idp_domain(self):\n        # create new idp\n        body = self.default_body.copy()\n        default_resp = self._create_default_idp(body=body)\n        default_idp = self._fetch_attribute_from_response(default_resp,\n                                                          \'identity_provider\')\n        idp_id = default_idp.get(\'id\')\n        self.assertIsNotNone(idp_id)\n        # create domain and try to update the idp\'s domain\n        domain = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(domain[\'id\'], domain)\n        body[\'domain_id\'] = domain[\'id\']\n        body = {\'identity_provider\': body}\n        url = self.base_url(suffix=idp_id)\n        self.patch(url, body=body, expected_status=http.client.BAD_REQUEST)\n\n    def test_create_idp_with_nonexistent_domain_id_fails(self):\n        body = self.default_body.copy()\n        body[\'description\'] = uuid.uuid4().hex\n        body[\'domain_id\'] = uuid.uuid4().hex\n        self._create_default_idp(body=body,\n                                 expected_status=http.client.NOT_FOUND)\n\n    def test_create_idp_remote(self):\n        """Create the IdentityProvider entity associated to remote_ids."""\n        keys_to_check = list(self.idp_keys)\n        keys_to_check.append(\'remote_ids\')\n        body = self.default_body.copy()\n        body[\'description\'] = uuid.uuid4().hex\n        body[\'remote_ids\'] = [uuid.uuid4().hex,\n                              uuid.uuid4().hex,\n                              uuid.uuid4().hex]\n        resp = self._create_default_idp(body=body)\n        self.assertValidResponse(resp, \'identity_provider\', dummy_validator,\n                                 keys_to_check=keys_to_check,\n                                 ref=body)\n        attr = self._fetch_attribute_from_response(resp, \'identity_provider\')\n        self.assertIdpDomainCreated(attr[\'id\'], attr[\'domain_id\'])\n\n    def test_create_idp_remote_repeated(self):\n        """Create two IdentityProvider entities with some remote_ids.\n\n        A remote_id is the same for both so the second IdP is not\n        created because of the uniqueness of the remote_ids\n\n        Expect HTTP 409 Conflict code for the latter call.\n\n        """\n        body = self.default_body.copy()\n        repeated_remote_id = uuid.uuid4().hex\n        body[\'remote_ids\'] = [uuid.uuid4().hex,\n                              uuid.uuid4().hex,\n                              uuid.uuid4().hex,\n                              repeated_remote_id]\n        self._create_default_idp(body=body)\n\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        body[\'remote_ids\'] = [uuid.uuid4().hex,\n                              repeated_remote_id]\n        resp = self.put(url, body={\'identity_provider\': body},\n                        expected_status=http.client.CONFLICT)\n\n        resp_data = jsonutils.loads(resp.body)\n        self.assertIn(\'Duplicate remote ID\',\n                      resp_data.get(\'error\', {}).get(\'message\'))\n\n    def test_create_idp_remote_empty(self):\n        """Create an IdP with empty remote_ids."""\n        keys_to_check = list(self.idp_keys)\n        keys_to_check.append(\'remote_ids\')\n        body = self.default_body.copy()\n        body[\'description\'] = uuid.uuid4().hex\n        body[\'remote_ids\'] = []\n        resp = self._create_default_idp(body=body)\n        self.assertValidResponse(resp, \'identity_provider\', dummy_validator,\n                                 keys_to_check=keys_to_check,\n                                 ref=body)\n\n    def test_create_idp_remote_none(self):\n        """Create an IdP with a None remote_ids."""\n        keys_to_check = list(self.idp_keys)\n        keys_to_check.append(\'remote_ids\')\n        body = self.default_body.copy()\n        body[\'description\'] = uuid.uuid4().hex\n        body[\'remote_ids\'] = None\n        resp = self._create_default_idp(body=body)\n        expected = body.copy()\n        expected[\'remote_ids\'] = []\n        self.assertValidResponse(resp, \'identity_provider\', dummy_validator,\n                                 keys_to_check=keys_to_check,\n                                 ref=expected)\n\n    def test_create_idp_authorization_ttl(self):\n        keys_to_check = list(self.idp_keys)\n        keys_to_check.append(\'authorization_ttl\')\n        body = self.default_body.copy()\n        body[\'description\'] = uuid.uuid4().hex\n        body[\'authorization_ttl\'] = 10080\n        resp = self._create_default_idp(body)\n        expected = body.copy()\n        self.assertValidResponse(resp, \'identity_provider\', dummy_validator,\n                                 keys_to_check=keys_to_check,\n                                 ref=expected)\n\n    def test_update_idp_remote_ids(self):\n        """Update IdP\'s remote_ids parameter."""\n        body = self.default_body.copy()\n        body[\'remote_ids\'] = [uuid.uuid4().hex]\n        default_resp = self._create_default_idp(body=body)\n        default_idp = self._fetch_attribute_from_response(default_resp,\n                                                          \'identity_provider\')\n        idp_id = default_idp.get(\'id\')\n        url = self.base_url(suffix=idp_id)\n        self.assertIsNotNone(idp_id)\n\n        body[\'remote_ids\'] = [uuid.uuid4().hex, uuid.uuid4().hex]\n\n        body = {\'identity_provider\': body}\n        resp = self.patch(url, body=body)\n        updated_idp = self._fetch_attribute_from_response(resp,\n                                                          \'identity_provider\')\n        body = body[\'identity_provider\']\n        self.assertEqual(sorted(body[\'remote_ids\']),\n                         sorted(updated_idp.get(\'remote_ids\')))\n\n        resp = self.get(url)\n        returned_idp = self._fetch_attribute_from_response(resp,\n                                                           \'identity_provider\')\n        self.assertEqual(sorted(body[\'remote_ids\']),\n                         sorted(returned_idp.get(\'remote_ids\')))\n\n    def test_update_idp_clean_remote_ids(self):\n        """Update IdP\'s remote_ids parameter with an empty list."""\n        body = self.default_body.copy()\n        body[\'remote_ids\'] = [uuid.uuid4().hex]\n        default_resp = self._create_default_idp(body=body)\n        default_idp = self._fetch_attribute_from_response(default_resp,\n                                                          \'identity_provider\')\n        idp_id = default_idp.get(\'id\')\n        url = self.base_url(suffix=idp_id)\n        self.assertIsNotNone(idp_id)\n\n        body[\'remote_ids\'] = []\n\n        body = {\'identity_provider\': body}\n        resp = self.patch(url, body=body)\n        updated_idp = self._fetch_attribute_from_response(resp,\n                                                          \'identity_provider\')\n        body = body[\'identity_provider\']\n        self.assertEqual(sorted(body[\'remote_ids\']),\n                         sorted(updated_idp.get(\'remote_ids\')))\n\n        resp = self.get(url)\n        returned_idp = self._fetch_attribute_from_response(resp,\n                                                           \'identity_provider\')\n        self.assertEqual(sorted(body[\'remote_ids\']),\n                         sorted(returned_idp.get(\'remote_ids\')))\n\n    def test_update_idp_remote_repeated(self):\n        """Update an IdentityProvider entity reusing a remote_id.\n\n        A remote_id is the same for both so the second IdP is not\n        updated because of the uniqueness of the remote_ids.\n\n        Expect HTTP 409 Conflict code for the latter call.\n\n        """\n        # Create first identity provider\n        body = self.default_body.copy()\n        repeated_remote_id = uuid.uuid4().hex\n        body[\'remote_ids\'] = [uuid.uuid4().hex, repeated_remote_id]\n        self._create_default_idp(body=body)\n\n        # Create second identity provider (without remote_ids)\n        body = self.default_body.copy()\n        default_resp = self._create_default_idp(body=body)\n        default_idp = self._fetch_attribute_from_response(default_resp,\n                                                          \'identity_provider\')\n        idp_id = default_idp.get(\'id\')\n        url = self.base_url(suffix=idp_id)\n\n        body[\'remote_ids\'] = [repeated_remote_id]\n        resp = self.patch(url, body={\'identity_provider\': body},\n                          expected_status=http.client.CONFLICT)\n        resp_data = jsonutils.loads(resp.body)\n        self.assertIn(\'Duplicate remote ID\',\n                      resp_data[\'error\'][\'message\'])\n\n    def test_update_idp_authorization_ttl(self):\n        body = self.default_body.copy()\n        body[\'authorization_ttl\'] = 10080\n        default_resp = self._create_default_idp(body=body)\n        default_idp = self._fetch_attribute_from_response(default_resp,\n                                                          \'identity_provider\')\n        idp_id = default_idp.get(\'id\')\n        url = self.base_url(suffix=idp_id)\n        self.assertIsNotNone(idp_id)\n\n        body[\'authorization_ttl\'] = None\n\n        body = {\'identity_provider\': body}\n        resp = self.patch(url, body=body)\n        updated_idp = self._fetch_attribute_from_response(resp,\n                                                          \'identity_provider\')\n        body = body[\'identity_provider\']\n        self.assertEqual(body[\'authorization_ttl\'],\n                         updated_idp.get(\'authorization_ttl\'))\n\n        resp = self.get(url)\n        returned_idp = self._fetch_attribute_from_response(resp,\n                                                           \'identity_provider\')\n        self.assertEqual(body[\'authorization_ttl\'],\n                         returned_idp.get(\'authorization_ttl\'))\n\n    def test_list_head_idps(self, iterations=5):\n        """List all available IdentityProviders.\n\n        This test collects ids of created IdPs and\n        intersects it with the list of all available IdPs.\n        List of all IdPs can be a superset of IdPs created in this test,\n        because other tests also create IdPs.\n\n        """\n        def get_id(resp):\n            r = self._fetch_attribute_from_response(resp,\n                                                    \'identity_provider\')\n            return r.get(\'id\')\n\n        ids = []\n        for _ in range(iterations):\n            id = get_id(self._create_default_idp())\n            ids.append(id)\n        ids = set(ids)\n\n        keys_to_check = self.idp_keys\n        keys_to_check.append(\'domain_id\')\n        url = self.base_url()\n        resp = self.get(url)\n        self.assertValidListResponse(resp, \'identity_providers\',\n                                     dummy_validator,\n                                     keys_to_check=keys_to_check)\n        entities = self._fetch_attribute_from_response(resp,\n                                                       \'identity_providers\')\n        entities_ids = set([e[\'id\'] for e in entities])\n        ids_intersection = entities_ids.intersection(ids)\n        self.assertEqual(ids_intersection, ids)\n\n        self.head(url, expected_status=http.client.OK)\n\n    def test_filter_list_head_idp_by_id(self):\n        def get_id(resp):\n            r = self._fetch_attribute_from_response(resp,\n                                                    \'identity_provider\')\n            return r.get(\'id\')\n\n        idp1_id = get_id(self._create_default_idp())\n        idp2_id = get_id(self._create_default_idp())\n\n        # list the IdP, should get two IdP.\n        url = self.base_url()\n        resp = self.get(url)\n        entities = self._fetch_attribute_from_response(resp,\n                                                       \'identity_providers\')\n        entities_ids = [e[\'id\'] for e in entities]\n        self.assertCountEqual(entities_ids, [idp1_id, idp2_id])\n\n        # filter the IdP by ID.\n        url = self.base_url() + \'?id=\' + idp1_id\n        resp = self.get(url)\n        filtered_service_list = resp.json[\'identity_providers\']\n        self.assertThat(filtered_service_list, matchers.HasLength(1))\n        self.assertEqual(idp1_id, filtered_service_list[0].get(\'id\'))\n\n        self.head(url, expected_status=http.client.OK)\n\n    def test_filter_list_head_idp_by_enabled(self):\n        def get_id(resp):\n            r = self._fetch_attribute_from_response(resp,\n                                                    \'identity_provider\')\n            return r.get(\'id\')\n\n        idp1_id = get_id(self._create_default_idp())\n\n        body = self.default_body.copy()\n        body[\'enabled\'] = False\n        idp2_id = get_id(self._create_default_idp(body=body))\n\n        # list the IdP, should get two IdP.\n        url = self.base_url()\n        resp = self.get(url)\n        entities = self._fetch_attribute_from_response(resp,\n                                                       \'identity_providers\')\n        entities_ids = [e[\'id\'] for e in entities]\n        self.assertCountEqual(entities_ids, [idp1_id, idp2_id])\n\n        # filter the IdP by \'enabled\'.\n        url = self.base_url() + \'?enabled=True\'\n        resp = self.get(url)\n        filtered_service_list = resp.json[\'identity_providers\']\n        self.assertThat(filtered_service_list, matchers.HasLength(1))\n        self.assertEqual(idp1_id, filtered_service_list[0].get(\'id\'))\n\n        self.head(url, expected_status=http.client.OK)\n\n    def test_check_idp_uniqueness(self):\n        """Add same IdP twice.\n\n        Expect HTTP 409 Conflict code for the latter call.\n\n        """\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        body = self._http_idp_input()\n        domain = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(domain[\'id\'], domain)\n        body[\'domain_id\'] = domain[\'id\']\n        self.put(url, body={\'identity_provider\': body},\n                 expected_status=http.client.CREATED)\n        resp = self.put(url, body={\'identity_provider\': body},\n                        expected_status=http.client.CONFLICT)\n\n        resp_data = jsonutils.loads(resp.body)\n        self.assertIn(\'Duplicate entry\',\n                      resp_data.get(\'error\', {}).get(\'message\'))\n\n    def test_get_head_idp(self):\n        """Create and later fetch IdP."""\n        body = self._http_idp_input()\n        domain = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(domain[\'id\'], domain)\n        body[\'domain_id\'] = domain[\'id\']\n        default_resp = self._create_default_idp(body=body)\n        default_idp = self._fetch_attribute_from_response(default_resp,\n                                                          \'identity_provider\')\n        idp_id = default_idp.get(\'id\')\n        url = self.base_url(suffix=idp_id)\n        resp = self.get(url)\n        # Strip keys out of `body` dictionary. This is done\n        # to be python 3 compatible\n        body_keys = list(body)\n        self.assertValidResponse(resp, \'identity_provider\',\n                                 dummy_validator, keys_to_check=body_keys,\n                                 ref=body)\n\n        self.head(url, expected_status=http.client.OK)\n\n    def test_get_nonexisting_idp(self):\n        """Fetch nonexisting IdP entity.\n\n        Expected HTTP 404 Not Found status code.\n\n        """\n        idp_id = uuid.uuid4().hex\n        self.assertIsNotNone(idp_id)\n\n        url = self.base_url(suffix=idp_id)\n        self.get(url, expected_status=http.client.NOT_FOUND)\n\n    def test_delete_existing_idp(self):\n        """Create and later delete IdP.\n\n        Expect HTTP 404 Not Found for the GET IdP call.\n        """\n        default_resp = self._create_default_idp()\n        default_idp = self._fetch_attribute_from_response(default_resp,\n                                                          \'identity_provider\')\n        idp_id = default_idp.get(\'id\')\n        self.assertIsNotNone(idp_id)\n        url = self.base_url(suffix=idp_id)\n        self.delete(url)\n        self.get(url, expected_status=http.client.NOT_FOUND)\n\n    def test_delete_idp_also_deletes_assigned_protocols(self):\n        """Deleting an IdP will delete its assigned protocol."""\n        # create default IdP\n        default_resp = self._create_default_idp()\n        default_idp = self._fetch_attribute_from_response(default_resp,\n                                                          \'identity_provider\')\n        idp_id = default_idp[\'id\']\n        protocol_id = uuid.uuid4().hex\n\n        url = self.base_url(suffix=\'%(idp_id)s/protocols/%(protocol_id)s\')\n        idp_url = self.base_url(suffix=idp_id)\n\n        # assign protocol to IdP\n        kwargs = {\'expected_status\': http.client.CREATED}\n        resp, idp_id, proto = self._assign_protocol_to_idp(\n            url=url,\n            idp_id=idp_id,\n            proto=protocol_id,\n            **kwargs)\n\n        # removing IdP will remove the assigned protocol as well\n        self.assertEqual(\n            1, len(PROVIDERS.federation_api.list_protocols(idp_id))\n        )\n        self.delete(idp_url)\n        self.get(idp_url, expected_status=http.client.NOT_FOUND)\n        self.assertEqual(\n            0, len(PROVIDERS.federation_api.list_protocols(idp_id))\n        )\n\n    def test_delete_nonexisting_idp(self):\n        """Delete nonexisting IdP.\n\n        Expect HTTP 404 Not Found for the GET IdP call.\n        """\n        idp_id = uuid.uuid4().hex\n        url = self.base_url(suffix=idp_id)\n        self.delete(url, expected_status=http.client.NOT_FOUND)\n\n    def test_update_idp_mutable_attributes(self):\n        """Update IdP\'s mutable parameters."""\n        default_resp = self._create_default_idp()\n        default_idp = self._fetch_attribute_from_response(default_resp,\n                                                          \'identity_provider\')\n        idp_id = default_idp.get(\'id\')\n        url = self.base_url(suffix=idp_id)\n        self.assertIsNotNone(idp_id)\n\n        _enabled = not default_idp.get(\'enabled\')\n        body = {\'remote_ids\': [uuid.uuid4().hex, uuid.uuid4().hex],\n                \'description\': uuid.uuid4().hex,\n                \'enabled\': _enabled}\n\n        body = {\'identity_provider\': body}\n        resp = self.patch(url, body=body)\n    def test_create_mapping_with_blacklist_and_whitelist(self):\n        """Test for adding whitelist and blacklist in the rule\n\n        Server should respond with HTTP 400 error upon discovering both\n        ``whitelist`` and ``blacklist`` keywords in the same rule.\n\n        """\n        url = self.MAPPING_URL + uuid.uuid4().hex\n                self.assertEqual(body[key], updated_idp.get(key))\n\n        resp = self.get(url)\n        updated_idp = self._fetch_attribute_from_response(resp,\n                                                          \'identity_provider\')\n        for key in body.keys():\n            if isinstance(body[key], list):\n                self.assertEqual(sorted(body[key]),\n                                 sorted(updated_idp.get(key)))\n            else:\n                self.assertEqual(body[key], updated_idp.get(key))\n\n    def test_update_idp_immutable_attributes(self):\n        """Update IdP\'s immutable parameters.\n\n        Expect HTTP BAD REQUEST.\n\n        """\n        default_resp = self._create_default_idp()\n        default_idp = self._fetch_attribute_from_response(default_resp,\n                                                          \'identity_provider\')\n        idp_id = default_idp.get(\'id\')\n        self.assertIsNotNone(idp_id)\n\n        body = self._http_idp_input()\n        body[\'id\'] = uuid.uuid4().hex\n        body[\'protocols\'] = [uuid.uuid4().hex, uuid.uuid4().hex]\n\n        url = self.base_url(suffix=idp_id)\n        self.patch(url, body={\'identity_provider\': body},\n                   expected_status=http.client.BAD_REQUEST)\n\n    def test_update_nonexistent_idp(self):\n        """Update nonexistent IdP.\n\n        Expect HTTP 404 Not Found code.\n\n        """\n        idp_id = uuid.uuid4().hex\n        url = self.base_url(suffix=idp_id)\n        body = self._http_idp_input()\n        body[\'enabled\'] = False\n        body = {\'identity_provider\': body}\n\n        self.patch(url, body=body, expected_status=http.client.NOT_FOUND)\n\n    def test_assign_protocol_to_idp(self):\n        """Assign a protocol to existing IdP."""\n        self._assign_protocol_to_idp(expected_status=http.client.CREATED)\n\n    def test_protocol_composite_pk(self):\n        """Test that Keystone can add two entities.\n\n        The entities have identical names, however, attached to different\n        IdPs.\n\n        1. Add IdP and assign it protocol with predefined name\n        2. Add another IdP and assign it a protocol with same name.\n\n        Expect HTTP 201 code\n\n        """\n        url = self.base_url(suffix=\'%(idp_id)s/protocols/%(protocol_id)s\')\n\n        kwargs = {\'expected_status\': http.client.CREATED}\n        self._assign_protocol_to_idp(proto=\'saml2\',\n                                     url=url, **kwargs)\n\n        self._assign_protocol_to_idp(proto=\'saml2\',\n                                     url=url, **kwargs)\n\n    def test_protocol_idp_pk_uniqueness(self):\n        """Test whether Keystone checks for unique idp/protocol values.\n\n        Add same protocol twice, expect Keystone to reject a latter call and\n        return HTTP 409 Conflict code.\n\n        """\n        url = self.base_url(suffix=\'%(idp_id)s/protocols/%(protocol_id)s\')\n\n        kwargs = {\'expected_status\': http.client.CREATED}\n        resp, idp_id, proto = self._assign_protocol_to_idp(proto=\'saml2\',\n                                                           url=url, **kwargs)\n        kwargs = {\'expected_status\': http.client.CONFLICT}\n        self._assign_protocol_to_idp(\n            idp_id=idp_id, proto=\'saml2\', validate=False, url=url, **kwargs\n        )\n\n    def test_assign_protocol_to_nonexistent_idp(self):\n        """Assign protocol to IdP that doesn\'t exist.\n\n        Expect HTTP 404 Not Found code.\n\n        """\n        idp_id = uuid.uuid4().hex\n        kwargs = {\'expected_status\': http.client.NOT_FOUND}\n        self._assign_protocol_to_idp(proto=\'saml2\',\n                                     idp_id=idp_id,\n                                     validate=False,\n                                     **kwargs)\n\n    def test_crud_protocol_without_protocol_id_in_url(self):\n        # NOTE(morgan): This test is redundant but is added to ensure\n        # the url routing error in bug 1817313 is explicitly covered.\n        # create a protocol, but do not put the ID in the URL\n        idp_id, _ = self._create_and_decapsulate_response()\n        mapping_id = uuid.uuid4().hex\n        self._create_mapping(mapping_id=mapping_id)\n        protocol = {\n            \'id\': uuid.uuid4().hex,\n            \'mapping_id\': mapping_id\n        }\n        with self.test_client() as c:\n            token = self.get_scoped_token()\n            # DELETE/PATCH/PUT on non-trailing `/` results in\n            # METHOD_NOT_ALLOWED\n            c.delete(\'/v3/OS-FEDERATION/identity_providers/%(idp_id)s\'\n                     \'/protocols\' % {\'idp_id\': idp_id},\n                     headers={\'X-Auth-Token\': token},\n                     expected_status_code=http.client.METHOD_NOT_ALLOWED)\n            c.patch(\'/v3/OS-FEDERATION/identity_providers/%(idp_id)s\'\n                    \'/protocols/\' % {\'idp_id\': idp_id},\n                    json={\'protocol\': protocol},\n                    headers={\'X-Auth-Token\': token},\n                    expected_status_code=http.client.METHOD_NOT_ALLOWED)\n            c.put(\'/v3/OS-FEDERATION/identity_providers/%(idp_id)s\'\n                  \'/protocols\' % {\'idp_id\': idp_id},\n                  json={\'protocol\': protocol},\n                  headers={\'X-Auth-Token\': token},\n                  expected_status_code=http.client.METHOD_NOT_ALLOWED)\n\n            # DELETE/PATCH/PUT should raise 405 with trailing \'/\', it is\n            # remapped to without the trailing \'/\' by the normalization\n            # middleware.\n            c.delete(\'/v3/OS-FEDERATION/identity_providers/%(idp_id)s\'\n                     \'/protocols/\' % {\'idp_id\': idp_id},\n                     headers={\'X-Auth-Token\': token},\n                     expected_status_code=http.client.METHOD_NOT_ALLOWED)\n            c.patch(\'/v3/OS-FEDERATION/identity_providers/%(idp_id)s\'\n                    \'/protocols/\' % {\'idp_id\': idp_id},\n                    json={\'protocol\': protocol},\n                    headers={\'X-Auth-Token\': token},\n                    expected_status_code=http.client.METHOD_NOT_ALLOWED)\n            c.put(\'/v3/OS-FEDERATION/identity_providers/%(idp_id)s\'\n                  \'/protocols/\' % {\'idp_id\': idp_id},\n                  json={\'protocol\': protocol},\n                  headers={\'X-Auth-Token\': token},\n                  expected_status_code=http.client.METHOD_NOT_ALLOWED)\n\n    def test_get_head_protocol(self):\n        """Create and later fetch protocol tied to IdP."""\n        resp, idp_id, proto = self._assign_protocol_to_idp(\n            expected_status=http.client.CREATED)\n        proto_id = self._fetch_attribute_from_response(resp, \'protocol\')[\'id\']\n        url = "%s/protocols/%s" % (idp_id, proto_id)\n        url = self.base_url(suffix=url)\n\n        resp = self.get(url)\n\n        reference = {\'id\': proto_id}\n        # Strip keys out of `body` dictionary. This is done\n        # to be python 3 compatible\n        reference_keys = list(reference)\n        self.assertValidResponse(resp, \'protocol\',\n                                 dummy_validator,\n                                 keys_to_check=reference_keys,\n                                 ref=reference)\n\n        self.head(url, expected_status=http.client.OK)\n\n    def test_list_head_protocols(self):\n        """Create set of protocols and later list them.\n\n        Compare input and output id sets.\n\n        """\n        resp, idp_id, proto = self._assign_protocol_to_idp(\n            expected_status=http.client.CREATED)\n        iterations = random.randint(0, 16)\n        protocol_ids = []\n        for _ in range(iterations):\n            resp, _, proto = self._assign_protocol_to_idp(\n                idp_id=idp_id,\n                expected_status=http.client.CREATED)\n            proto_id = self._fetch_attribute_from_response(resp, \'protocol\')\n            proto_id = proto_id[\'id\']\n            protocol_ids.append(proto_id)\n\n        url = "%s/protocols" % idp_id\n        url = self.base_url(suffix=url)\n        resp = self.get(url)\n        self.assertValidListResponse(resp, \'protocols\',\n                                     dummy_validator,\n                                     keys_to_check=[\'id\'])\n        entities = self._fetch_attribute_from_response(resp, \'protocols\')\n        entities = set([entity[\'id\'] for entity in entities])\n        protocols_intersection = entities.intersection(protocol_ids)\n        self.assertEqual(protocols_intersection, set(protocol_ids))\n\n        self.head(url, expected_status=http.client.OK)\n\n    def test_update_protocols_attribute(self):\n        """Update protocol\'s attribute."""\n        resp, idp_id, proto = self._assign_protocol_to_idp(\n            expected_status=http.client.CREATED)\n        new_mapping_id = uuid.uuid4().hex\n        self._create_mapping(mapping_id=new_mapping_id)\n\n        url = "%s/protocols/%s" % (idp_id, proto)\n        url = self.base_url(suffix=url)\n        body = {\'mapping_id\': new_mapping_id}\n        resp = self.patch(url, body={\'protocol\': body})\n        self.assertValidResponse(resp, \'protocol\', dummy_validator,\n                                 keys_to_check=[\'id\', \'mapping_id\'],\n                                 ref={\'id\': proto,\n                                      \'mapping_id\': new_mapping_id}\n                                 )\n\n    def test_delete_protocol(self):\n        """Delete protocol.\n\n        Expect HTTP 404 Not Found code for the GET call after the protocol is\n        deleted.\n\n        """\n        url = self.base_url(suffix=\'%(idp_id)s/\'\n                                   \'protocols/%(protocol_id)s\')\n        resp, idp_id, proto = self._assign_protocol_to_idp(\n            expected_status=http.client.CREATED)\n        url = url % {\'idp_id\': idp_id,\n                     \'protocol_id\': proto}\n        self.delete(url)\n        self.get(url, expected_status=http.client.NOT_FOUND)\n\n\nclass MappingCRUDTests(test_v3.RestfulTestCase):\n    """A class for testing CRUD operations for Mappings."""\n\n    MAPPING_URL = \'/OS-FEDERATION/mappings/\'\n\n    def assertValidMappingListResponse(self, resp, *args, **kwargs):\n        return self.assertValidListResponse(\n            resp,\n            \'mappings\',\n            self.assertValidMapping,\n            keys_to_check=[],\n            *args,\n            **kwargs)\n\n    def assertValidMappingResponse(self, resp, *args, **kwargs):\n        return self.assertValidResponse(\n            resp,\n            \'mapping\',\n            self.assertValidMapping,\n            keys_to_check=[],\n            *args,\n            **kwargs)\n\n    def assertValidMapping(self, entity, ref=None):\n        self.assertIsNotNone(entity.get(\'id\'))\n        self.assertIsNotNone(entity.get(\'rules\'))\n        if ref:\n            self.assertEqual(entity[\'rules\'], ref[\'rules\'])\n        return entity\n\n    def _create_default_mapping_entry(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        resp = self.put(url,\n                        body={\'mapping\': mapping_fixtures.MAPPING_LARGE},\n                        expected_status=http.client.CREATED)\n        return resp\n\n    def _get_id_from_response(self, resp):\n        r = resp.result.get(\'mapping\')\n        return r.get(\'id\')\n\n    def test_mapping_create(self):\n        resp = self._create_default_mapping_entry()\n        self.assertValidMappingResponse(resp, mapping_fixtures.MAPPING_LARGE)\n\n    def test_mapping_list_head(self):\n        url = self.MAPPING_URL\n        self._create_default_mapping_entry()\n        resp = self.get(url)\n        entities = resp.result.get(\'mappings\')\n        self.assertIsNotNone(entities)\n        self.assertResponseStatus(resp, http.client.OK)\n        self.assertValidListLinks(resp.result.get(\'links\'))\n        self.assertEqual(1, len(entities))\n        self.head(url, expected_status=http.client.OK)\n\n    def test_mapping_delete(self):\n        url = self.MAPPING_URL + \'%(mapping_id)s\'\n        resp = self._create_default_mapping_entry()\n        mapping_id = self._get_id_from_response(resp)\n        url = url % {\'mapping_id\': str(mapping_id)}\n        resp = self.delete(url)\n        self.assertResponseStatus(resp, http.client.NO_CONTENT)\n        self.get(url, expected_status=http.client.NOT_FOUND)\n\n    def test_mapping_get_head(self):\n        url = self.MAPPING_URL + \'%(mapping_id)s\'\n        resp = self._create_default_mapping_entry()\n        mapping_id = self._get_id_from_response(resp)\n        url = url % {\'mapping_id\': mapping_id}\n        resp = self.get(url)\n        self.assertValidMappingResponse(resp, mapping_fixtures.MAPPING_LARGE)\n        self.head(url, expected_status=http.client.OK)\n\n    def test_mapping_update(self):\n        url = self.MAPPING_URL + \'%(mapping_id)s\'\n        resp = self._create_default_mapping_entry()\n        mapping_id = self._get_id_from_response(resp)\n        url = url % {\'mapping_id\': mapping_id}\n        resp = self.patch(url,\n                          body={\'mapping\': mapping_fixtures.MAPPING_SMALL})\n        self.assertValidMappingResponse(resp, mapping_fixtures.MAPPING_SMALL)\n        resp = self.get(url)\n        self.assertValidMappingResponse(resp, mapping_fixtures.MAPPING_SMALL)\n\n    def test_delete_mapping_dne(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.delete(url, expected_status=http.client.NOT_FOUND)\n\n    def test_get_mapping_dne(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.get(url, expected_status=http.client.NOT_FOUND)\n\n    def test_create_mapping_bad_requirements(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': mapping_fixtures.MAPPING_BAD_REQ})\n\n    def test_create_mapping_no_rules(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': mapping_fixtures.MAPPING_NO_RULES})\n\n    def test_create_mapping_no_remote_objects(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': mapping_fixtures.MAPPING_NO_REMOTE})\n\n    def test_create_mapping_bad_value(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': mapping_fixtures.MAPPING_BAD_VALUE})\n\n    def test_create_mapping_missing_local(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': mapping_fixtures.MAPPING_MISSING_LOCAL})\n\n    def test_create_mapping_missing_type(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': mapping_fixtures.MAPPING_MISSING_TYPE})\n\n    def test_create_mapping_wrong_type(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': mapping_fixtures.MAPPING_WRONG_TYPE})\n\n    def test_create_mapping_extra_remote_properties_not_any_of(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        mapping = mapping_fixtures.MAPPING_EXTRA_REMOTE_PROPS_NOT_ANY_OF\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': mapping})\n\n    def test_create_mapping_extra_remote_properties_any_one_of(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        mapping = mapping_fixtures.MAPPING_EXTRA_REMOTE_PROPS_ANY_ONE_OF\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': mapping})\n\n    def test_create_mapping_extra_remote_properties_just_type(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        mapping = mapping_fixtures.MAPPING_EXTRA_REMOTE_PROPS_JUST_TYPE\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': mapping})\n\n    def test_create_mapping_empty_map(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': {}})\n\n    def test_create_mapping_extra_rules_properties(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': mapping_fixtures.MAPPING_EXTRA_RULES_PROPS})\n\n    def test_create_mapping_with_blacklist_and_whitelist(self):\n        """Test for adding whitelist and blacklist in the rule.\n\n        Server should respond with HTTP 400 Bad Request error upon discovering\n        both ``whitelist`` and ``blacklist`` keywords in the same rule.\n\n        """\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        mapping = mapping_fixtures.MAPPING_GROUPS_WHITELIST_AND_BLACKLIST\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': mapping})\n\n    def test_create_mapping_with_local_user_and_local_domain(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        resp = self.put(\n            url,\n            body={\n                \'mapping\': mapping_fixtures.MAPPING_LOCAL_USER_LOCAL_DOMAIN\n            },\n            expected_status=http.client.CREATED)\n        self.assertValidMappingResponse(\n            resp, mapping_fixtures.MAPPING_LOCAL_USER_LOCAL_DOMAIN)\n\n    def test_create_mapping_with_ephemeral(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        resp = self.put(\n            url,\n            body={\'mapping\': mapping_fixtures.MAPPING_EPHEMERAL_USER},\n            expected_status=http.client.CREATED)\n        self.assertValidMappingResponse(\n            resp, mapping_fixtures.MAPPING_EPHEMERAL_USER)\n\n    def test_create_mapping_with_bad_user_type(self):\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        # get a copy of a known good map\n        bad_mapping = copy.deepcopy(mapping_fixtures.MAPPING_EPHEMERAL_USER)\n        # now sabotage the user type\n        bad_mapping[\'rules\'][0][\'local\'][0][\'user\'][\'type\'] = uuid.uuid4().hex\n        self.put(url, expected_status=http.client.BAD_REQUEST,\n                 body={\'mapping\': bad_mapping})\n\n    def test_create_shadow_mapping_without_roles_fails(self):\n        """Validate that mappings with projects contain roles when created."""\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.put(\n            url,\n            body={\'mapping\': mapping_fixtures.MAPPING_PROJECTS_WITHOUT_ROLES},\n            expected_status=http.client.BAD_REQUEST\n        )\n\n    def test_update_shadow_mapping_without_roles_fails(self):\n        """Validate that mappings with projects contain roles when updated."""\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        resp = self.put(\n            url,\n            body={\'mapping\': mapping_fixtures.MAPPING_PROJECTS},\n            expected_status=http.client.CREATED\n        )\n        self.assertValidMappingResponse(\n            resp, mapping_fixtures.MAPPING_PROJECTS\n        )\n        self.patch(\n            url,\n            body={\'mapping\': mapping_fixtures.MAPPING_PROJECTS_WITHOUT_ROLES},\n            expected_status=http.client.BAD_REQUEST\n        )\n\n    def test_create_shadow_mapping_without_name_fails(self):\n        """Validate project mappings contain the project name when created."""\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        self.put(\n            url,\n            body={\'mapping\': mapping_fixtures.MAPPING_PROJECTS_WITHOUT_NAME},\n            expected_status=http.client.BAD_REQUEST\n        )\n\n    def test_update_shadow_mapping_without_name_fails(self):\n        """Validate project mappings contain the project name when updated."""\n        url = self.MAPPING_URL + uuid.uuid4().hex\n        resp = self.put(\n            url,\n            body={\'mapping\': mapping_fixtures.MAPPING_PROJECTS},\n            expected_status=http.client.CREATED\n        )\n        self.assertValidMappingResponse(\n            resp, mapping_fixtures.MAPPING_PROJECTS\n        )\n        self.patch(\n            url,\n            body={\'mapping\': mapping_fixtures.MAPPING_PROJECTS_WITHOUT_NAME},\n            expected_status=http.client.BAD_REQUEST\n        )\n\n\nclass FederatedTokenTests(test_v3.RestfulTestCase, FederatedSetupMixin):\n\n    def auth_plugin_config_override(self):\n        methods = [\'saml2\', \'token\']\n        super(FederatedTokenTests, self).auth_plugin_config_override(methods)\n\n    def setUp(self):\n        super(FederatedTokenTests, self).setUp()\n        self._notifications = []\n\n        def fake_saml_notify(action, user_id, group_ids,\n                             identity_provider, protocol, token_id, outcome):\n            note = {\n                \'action\': action,\n                \'user_id\': user_id,\n                \'identity_provider\': identity_provider,\n                \'protocol\': protocol,\n                \'send_notification_called\': True}\n            self._notifications.append(note)\n\n        self.useFixture(fixtures.MockPatchObject(\n            notifications,\n            \'send_saml_audit_notification\',\n            fake_saml_notify))\n\n    def _assert_last_notify(self, action, identity_provider, protocol,\n                            user_id=None):\n        self.assertTrue(self._notifications)\n        note = self._notifications[-1]\n        if user_id:\n            self.assertEqual(note[\'user_id\'], user_id)\n        self.assertEqual(note[\'action\'], action)\n        self.assertEqual(note[\'identity_provider\'], identity_provider)\n        self.assertEqual(note[\'protocol\'], protocol)\n        self.assertTrue(note[\'send_notification_called\'])\n\n    def load_fixtures(self, fixtures):\n        super(FederatedTokenTests, self).load_fixtures(fixtures)\n        self.load_federation_sample_data()\n\n    def test_issue_unscoped_token_notify(self):\n        self._issue_unscoped_token()\n        self._assert_last_notify(self.ACTION, self.IDP, self.PROTOCOL)\n\n    def test_issue_unscoped_token(self):\n        r = self._issue_unscoped_token()\n        token_resp = render_token.render_token_response_from_model(r)[\'token\']\n        self.assertValidMappedUser(token_resp)\n\n    def test_default_domain_scoped_token(self):\n        # Make sure federated users can get tokens scoped to the default\n        # domain, which has a non-uuid ID by default (e.g., `default`). We want\n        # to make sure the token provider handles string types properly if the\n        # ID isn\'t compressed into byte format during validation. Turn off\n        # cache on issue so that we validate the token online right after we\n        # get it to make sure the token provider is called.\n        self.config_fixture.config(group=\'token\', cache_on_issue=False)\n\n        # Grab an unscoped token to get a domain-scoped token with.\n        token = self._issue_unscoped_token()\n\n        # Give the user a direct role assignment on the default domain, so they\n        # can get a federated domain-scoped token.\n        PROVIDERS.assignment_api.create_grant(\n            self.role_admin[\'id\'], user_id=token.user_id,\n            domain_id=CONF.identity.default_domain_id\n        )\n\n        # Get a token scoped to the default domain with an ID of `default`,\n        # which isn\'t a uuid type, but we should be able to handle it\n        # accordingly in the token formatters/providers.\n        auth_request = {\n            \'auth\': {\n                \'identity\': {\n                    \'methods\': [\n                        \'token\'\n                    ],\n                    \'token\': {\n                        \'id\': token.id\n                    }\n                },\n                \'scope\': {\n                    \'domain\': {\n                        \'id\': CONF.identity.default_domain_id\n                    }\n                }\n            }\n        }\n        r = self.v3_create_token(auth_request)\n        domain_scoped_token_id = r.headers.get(\'X-Subject-Token\')\n\n        # Validate the token to make sure the token providers handle non-uuid\n        # domain IDs properly.\n        headers = {\'X-Subject-Token\': domain_scoped_token_id}\n        self.get(\n            \'/auth/tokens\',\n            token=domain_scoped_token_id,\n            headers=headers\n        )\n\n    def test_issue_the_same_unscoped_token_with_user_deleted(self):\n        r = self._issue_unscoped_token()\n        token = render_token.render_token_response_from_model(r)[\'token\']\n        user1 = token[\'user\']\n        user_id1 = user1.pop(\'id\')\n\n        # delete the referenced user, and authenticate again. Keystone should\n        # create another new shadow user.\n        PROVIDERS.identity_api.delete_user(user_id1)\n\n        r = self._issue_unscoped_token()\n        token = render_token.render_token_response_from_model(r)[\'token\']\n        user2 = token[\'user\']\n        user_id2 = user2.pop(\'id\')\n\n        # Only the user_id is different. Other properties include\n        # identity_provider, protocol, groups and domain are the same.\n        self.assertIsNot(user_id2, user_id1)\n        self.assertEqual(user1, user2)\n\n    def test_issue_unscoped_token_disabled_idp(self):\n        """Check if authentication works with disabled identity providers.\n\n        Test plan:\n        1) Disable default IdP\n        2) Try issuing unscoped token for that IdP\n        3) Expect server to forbid authentication\n\n        """\n        enabled_false = {\'enabled\': False}\n        PROVIDERS.federation_api.update_idp(self.IDP, enabled_false)\n        self.assertRaises(exception.Forbidden,\n                          self._issue_unscoped_token)\n\n    def test_issue_unscoped_token_group_names_in_mapping(self):\n        r = self._issue_unscoped_token(assertion=\'ANOTHER_CUSTOMER_ASSERTION\')\n        ref_groups = set([self.group_customers[\'id\'], self.group_admins[\'id\']])\n        token_groups = r.federated_groups\n        token_groups = set([group[\'id\'] for group in token_groups])\n        self.assertEqual(ref_groups, token_groups)\n\n    def test_issue_unscoped_tokens_nonexisting_group(self):\n        self._issue_unscoped_token(assertion=\'ANOTHER_TESTER_ASSERTION\')\n\n    def test_issue_unscoped_token_with_remote_no_attribute(self):\n        self._issue_unscoped_token(idp=self.IDP_WITH_REMOTE,\n                                   environment={\n                                       self.REMOTE_ID_ATTR:\n                                           self.REMOTE_IDS[0]\n                                   })\n\n    def test_issue_unscoped_token_with_remote(self):\n        self.config_fixture.config(group=\'federation\',\n                                   remote_id_attribute=self.REMOTE_ID_ATTR)\n        self._issue_unscoped_token(idp=self.IDP_WITH_REMOTE,\n                                   environment={\n                                       self.REMOTE_ID_ATTR:\n                                           self.REMOTE_IDS[0]\n                                   })\n\n    def test_issue_unscoped_token_with_saml2_remote(self):\n        self.config_fixture.config(group=\'saml2\',\n                                   remote_id_attribute=self.REMOTE_ID_ATTR)\n        self._issue_unscoped_token(idp=self.IDP_WITH_REMOTE,\n                                   environment={\n                                       self.REMOTE_ID_ATTR:\n                                           self.REMOTE_IDS[0]\n                                   })\n\n    def test_issue_unscoped_token_with_remote_different(self):\n        self.config_fixture.config(group=\'federation\',\n                                   remote_id_attribute=self.REMOTE_ID_ATTR)\n        self.assertRaises(exception.Forbidden,\n                          self._issue_unscoped_token,\n                          idp=self.IDP_WITH_REMOTE,\n                          environment={\n                              self.REMOTE_ID_ATTR: uuid.uuid4().hex\n                          })\n\n    def test_issue_unscoped_token_with_remote_default_overwritten(self):\n        """Test that protocol remote_id_attribute has higher priority.\n\n        Make sure the parameter stored under ``protocol`` section has higher\n        priority over parameter from default ``federation`` configuration\n        section.\n\n        """\n        self.config_fixture.config(group=\'saml2\',\n                                   remote_id_attribute=self.REMOTE_ID_ATTR)\n        self.config_fixture.config(group=\'federation\',\n                                   remote_id_attribute=uuid.uuid4().hex)\n        self._issue_unscoped_token(idp=self.IDP_WITH_REMOTE,\n                                   environment={\n                                       self.REMOTE_ID_ATTR:\n                                           self.REMOTE_IDS[0]\n                                   })\n\n    def test_issue_unscoped_token_with_remote_unavailable(self):\n        self.config_fixture.config(group=\'federation\',\n                                   remote_id_attribute=self.REMOTE_ID_ATTR)\n        self.assertRaises(exception.Unauthorized,\n                          self._issue_unscoped_token,\n                          idp=self.IDP_WITH_REMOTE,\n                          environment={\n                              uuid.uuid4().hex: uuid.uuid4().hex\n                          })\n\n    def test_issue_unscoped_token_with_remote_user_as_empty_string(self):\n        # make sure that REMOTE_USER set as the empty string won\'t interfere\n        self._issue_unscoped_token(environment={\'REMOTE_USER\': \'\'})\n\n    def test_issue_unscoped_token_no_groups(self):\n        r = self._issue_unscoped_token(assertion=\'USER_NO_GROUPS_ASSERTION\')\n        token_groups = r.federated_groups\n        self.assertEqual(0, len(token_groups))\n\n    def test_issue_scoped_token_no_groups(self):\n        """Verify that token without groups cannot get scoped to project.\n\n        This test is required because of bug 1677723.\n        """\n        # issue unscoped token with no groups\n        r = self._issue_unscoped_token(assertion=\'USER_NO_GROUPS_ASSERTION\')\n        token_groups = r.federated_groups\n        self.assertEqual(0, len(token_groups))\n        unscoped_token = r.id\n\n        # let admin get roles in a project\n        self.proj_employees\n        admin = unit.new_user_ref(CONF.identity.default_domain_id)\n        PROVIDERS.identity_api.create_user(admin)\n        PROVIDERS.assignment_api.create_grant(\n            self.role_admin[\'id\'], user_id=admin[\'id\'],\n            project_id=self.proj_employees[\'id\']\n        )\n\n        # try to scope the token. It should fail\n        scope = self._scope_request(\n            unscoped_token, \'project\', self.proj_employees[\'id\']\n        )\n        self.v3_create_token(\n            scope, expected_status=http.client.UNAUTHORIZED)\n\n    def test_issue_unscoped_token_malformed_environment(self):\n        """Test whether non string objects are filtered out.\n\n        Put non string objects into the environment, inject\n        correct assertion and try to get an unscoped token.\n        Expect server not to fail on using split() method on\n        non string objects and return token id in the HTTP header.\n\n        """\n        environ = {\n            \'malformed_object\': object(),\n            \'another_bad_idea\': tuple(range(10)),\n            \'yet_another_bad_param\': dict(zip(uuid.uuid4().hex, range(32)))\n        }\n        environ.update(mapping_fixtures.EMPLOYEE_ASSERTION)\n        with self.make_request(environ=environ):\n            authentication.authenticate_for_token(self.UNSCOPED_V3_SAML2_REQ)\n\n    def test_scope_to_project_once_notify(self):\n        r = self.v3_create_token(\n            self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_EMPLOYEE)\n        user_id = r.json[\'token\'][\'user\'][\'id\']\n        self._assert_last_notify(self.ACTION, self.IDP, self.PROTOCOL, user_id)\n\n    def test_scope_to_project_once(self):\n        r = self.v3_create_token(\n            self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_EMPLOYEE)\n        token_resp = r.result[\'token\']\n        project_id = token_resp[\'project\'][\'id\']\n        self._check_project_scoped_token_attributes(token_resp, project_id)\n        roles_ref = [self.role_employee]\n\n        projects_ref = self.proj_employees\n        self._check_projects_and_roles(token_resp, roles_ref, projects_ref)\n        self.assertValidMappedUser(token_resp)\n\n    def test_scope_token_with_idp_disabled(self):\n        """Scope token issued by disabled IdP.\n\n        Try scoping the token issued by an IdP which is disabled now. Expect\n        server to refuse scoping operation.\n\n        This test confirms correct behaviour when IdP was enabled and unscoped\n        token was issued, but disabled before user tries to scope the token.\n        Here we assume the unscoped token was already issued and start from\n        the moment where IdP is being disabled and unscoped token is being\n        used.\n\n        Test plan:\n        1) Disable IdP\n        2) Try scoping unscoped token\n\n        """\n        enabled_false = {\'enabled\': False}\n        PROVIDERS.federation_api.update_idp(self.IDP, enabled_false)\n        self.v3_create_token(\n            self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_CUSTOMER,\n            expected_status=http.client.FORBIDDEN)\n\n    def test_validate_token_after_deleting_idp_raises_not_found(self):\n        token = self.v3_create_token(\n            self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN\n        )\n        token_id = token.headers.get(\'X-Subject-Token\')\n        federated_info = token.json_body[\'token\'][\'user\'][\'OS-FEDERATION\']\n        idp_id = federated_info[\'identity_provider\'][\'id\']\n        PROVIDERS.federation_api.delete_idp(idp_id)\n        headers = {\n            \'X-Subject-Token\': token_id\n        }\n        # NOTE(lbragstad): This raises a 404 NOT FOUND because the identity\n        # provider is no longer present. We raise 404 NOT FOUND when we\n        # validate a token and a project or domain no longer exists.\n        self.get(\n            \'/auth/tokens/\',\n            token=token_id,\n            headers=headers,\n            expected_status=http.client.NOT_FOUND\n        )\n\n    def test_deleting_idp_cascade_deleting_fed_user(self):\n        token = self.v3_create_token(\n            self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN\n        )\n        federated_info = token.json_body[\'token\'][\'user\'][\'OS-FEDERATION\']\n        idp_id = federated_info[\'identity_provider\'][\'id\']\n\n        # There are three fed users (from \'EMPLOYEE_ASSERTION\',\n        # \'CUSTOMER_ASSERTION\', \'ADMIN_ASSERTION\') with the specified idp.\n        hints = driver_hints.Hints()\n        hints.add_filter(\'idp_id\', idp_id)\n        fed_users = PROVIDERS.shadow_users_api.get_federated_users(hints)\n        self.assertEqual(3, len(fed_users))\n        idp_domain_id = PROVIDERS.federation_api.get_idp(idp_id)[\'domain_id\']\n        for fed_user in fed_users:\n            self.assertEqual(idp_domain_id, fed_user[\'domain_id\'])\n\n        # Delete the idp\n        PROVIDERS.federation_api.delete_idp(idp_id)\n\n        # The related federated user should be deleted as well.\n        hints = driver_hints.Hints()\n        hints.add_filter(\'idp_id\', idp_id)\n        fed_users = PROVIDERS.shadow_users_api.get_federated_users(hints)\n        self.assertEqual([], fed_users)\n\n    def test_scope_to_bad_project(self):\n        """Scope unscoped token with a project we don\'t have access to."""\n        self.v3_create_token(\n            self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_CUSTOMER,\n            expected_status=http.client.UNAUTHORIZED)\n\n    def test_scope_to_project_multiple_times(self):\n        """Try to scope the unscoped token multiple times.\n\n        The new tokens should be scoped to:\n\n        * Customers\' project\n        * Employees\' project\n\n        """\n        bodies = (self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN,\n                  self.TOKEN_SCOPE_PROJECT_CUSTOMER_FROM_ADMIN)\n        project_ids = (self.proj_employees[\'id\'],\n                       self.proj_customers[\'id\'])\n        for body, project_id_ref in zip(bodies, project_ids):\n            r = self.v3_create_token(body)\n            token_resp = r.result[\'token\']\n            self._check_project_scoped_token_attributes(token_resp,\n                                                        project_id_ref)\n\n    def test_scope_to_project_with_duplicate_roles_returns_single_role(self):\n        r = self.v3_create_token(self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN)\n\n        # Even though the process of obtaining a token shows that there is a\n        # role assignment on a project, we should attempt to create a duplicate\n        # assignment somewhere. Do this by creating a direct role assignment\n        # with each role against the project the token was scoped to.\n        user_id = r.json_body[\'token\'][\'user\'][\'id\']\n        project_id = r.json_body[\'token\'][\'project\'][\'id\']\n        for role in r.json_body[\'token\'][\'roles\']:\n            PROVIDERS.assignment_api.create_grant(\n                role_id=role[\'id\'], user_id=user_id, project_id=project_id\n            )\n\n        # Ensure all roles in the token are unique even though we know there\n        # should be duplicate role assignment from the assertions and the\n        # direct role assignments we just created.\n        r = self.v3_create_token(self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN)\n        known_role_ids = []\n        for role in r.json_body[\'token\'][\'roles\']:\n            self.assertNotIn(role[\'id\'], known_role_ids)\n            known_role_ids.append(role[\'id\'])\n\n    def test_scope_to_project_with_only_inherited_roles(self):\n        """Try to scope token whose only roles are inherited."""\n        r = self.v3_create_token(\n            self.TOKEN_SCOPE_PROJECT_INHERITED_FROM_CUSTOMER)\n        token_resp = r.result[\'token\']\n        self._check_project_scoped_token_attributes(\n            token_resp, self.project_inherited[\'id\'])\n        roles_ref = [self.role_customer]\n        projects_ref = self.project_inherited\n        self._check_projects_and_roles(token_resp, roles_ref, projects_ref)\n        self.assertValidMappedUser(token_resp)\n\n    def test_scope_token_from_nonexistent_unscoped_token(self):\n        """Try to scope token from non-existent unscoped token."""\n        self.v3_create_token(\n            self.TOKEN_SCOPE_PROJECT_FROM_NONEXISTENT_TOKEN,\n            expected_status=http.client.NOT_FOUND)\n\n    def test_issue_token_from_rules_without_user(self):\n        environ = copy.deepcopy(mapping_fixtures.BAD_TESTER_ASSERTION)\n        with self.make_request(environ=environ):\n            self.assertRaises(exception.Unauthorized,\n                              authentication.authenticate_for_token,\n                              self.UNSCOPED_V3_SAML2_REQ)\n\n    def test_issue_token_with_nonexistent_group(self):\n        """Inject assertion that matches rule issuing bad group id.\n\n        Expect server to find out that some groups are missing in the\n        backend and raise exception.MappedGroupNotFound exception.\n\n        """\n        self.assertRaises(exception.MappedGroupNotFound,\n                          self._issue_unscoped_token,\n                          assertion=\'CONTRACTOR_ASSERTION\')\n\n    def test_scope_to_domain_once(self):\n        r = self.v3_create_token(self.TOKEN_SCOPE_DOMAIN_A_FROM_CUSTOMER)\n        token_resp = r.result[\'token\']\n        self._check_domain_scoped_token_attributes(token_resp,\n                                                   self.domainA[\'id\'])\n\n    def test_scope_to_domain_multiple_tokens(self):\n        """Issue multiple tokens scoping to different domains.\n\n        The new tokens should be scoped to:\n\n        * domainA\n        * domainB\n        * domainC\n\n        """\n        bodies = (self.TOKEN_SCOPE_DOMAIN_A_FROM_ADMIN,\n                  self.TOKEN_SCOPE_DOMAIN_B_FROM_ADMIN,\n                  self.TOKEN_SCOPE_DOMAIN_C_FROM_ADMIN)\n        domain_ids = (self.domainA[\'id\'],\n                      self.domainB[\'id\'],\n                      self.domainC[\'id\'])\n\n        for body, domain_id_ref in zip(bodies, domain_ids):\n            r = self.v3_create_token(body)\n            token_resp = r.result[\'token\']\n            self._check_domain_scoped_token_attributes(token_resp,\n                                                       domain_id_ref)\n\n    def test_scope_to_domain_with_only_inherited_roles_fails(self):\n        """Try to scope to a domain that has no direct roles."""\n        self.v3_create_token(\n            self.TOKEN_SCOPE_DOMAIN_D_FROM_CUSTOMER,\n            expected_status=http.client.UNAUTHORIZED)\n\n    def test_list_projects(self):\n        urls = (\'/OS-FEDERATION/projects\', \'/auth/projects\')\n\n        token = (self.tokens[\'CUSTOMER_ASSERTION\'],\n                 self.tokens[\'EMPLOYEE_ASSERTION\'],\n                 self.tokens[\'ADMIN_ASSERTION\'])\n\n        projects_refs = (set([self.proj_customers[\'id\'],\n                              self.project_inherited[\'id\']]),\n                         set([self.proj_employees[\'id\'],\n                              self.project_all[\'id\']]),\n                         set([self.proj_employees[\'id\'],\n                              self.project_all[\'id\'],\n                              self.proj_customers[\'id\'],\n                              self.project_inherited[\'id\']]))\n\n        for token, projects_ref in zip(token, projects_refs):\n            for url in urls:\n                r = self.get(url, token=token)\n                projects_resp = r.result[\'projects\']\n                projects = set(p[\'id\'] for p in projects_resp)\n                self.assertEqual(projects_ref, projects,\n                                 \'match failed for url %s\' % url)\n\n    # TODO(samueldmq): Create another test class for role inheritance tests.\n    # The advantage would be to reduce the complexity of this test class and\n    # have tests specific to this functionality grouped, easing readability and\n    # maintenability.\n    def test_list_projects_for_inherited_project_assignment(self):\n        # Create a subproject\n        subproject_inherited = unit.new_project_ref(\n            domain_id=self.domainD[\'id\'],\n            parent_id=self.project_inherited[\'id\'])\n        PROVIDERS.resource_api.create_project(\n            subproject_inherited[\'id\'], subproject_inherited\n        )\n\n        # Create an inherited role assignment\n        PROVIDERS.assignment_api.create_grant(\n            role_id=self.role_employee[\'id\'],\n            group_id=self.group_employees[\'id\'],\n            project_id=self.project_inherited[\'id\'],\n            inherited_to_projects=True)\n\n        # Define expected projects from employee assertion, which contain\n        # the created subproject\n        expected_project_ids = [self.project_all[\'id\'],\n                                self.proj_employees[\'id\'],\n                                subproject_inherited[\'id\']]\n\n        # Assert expected projects for both available URLs\n        for url in (\'/OS-FEDERATION/projects\', \'/auth/projects\'):\n            r = self.get(url, token=self.tokens[\'EMPLOYEE_ASSERTION\'])\n            project_ids = [project[\'id\'] for project in r.result[\'projects\']]\n\n            self.assertEqual(len(expected_project_ids), len(project_ids))\n            for expected_project_id in expected_project_ids:\n                self.assertIn(expected_project_id, project_ids,\n                              \'Projects match failed for url %s\' % url)\n\n    def test_list_domains(self):\n        urls = (\'/OS-FEDERATION/domains\', \'/auth/domains\')\n\n        tokens = (self.tokens[\'CUSTOMER_ASSERTION\'],\n                  self.tokens[\'EMPLOYEE_ASSERTION\'],\n                  self.tokens[\'ADMIN_ASSERTION\'])\n\n        # NOTE(henry-nash): domain D does not appear in the expected results\n        # since it only had inherited roles (which only apply to projects\n        # within the domain)\n\n        domain_refs = (set([self.domainA[\'id\']]),\n                       set([self.domainA[\'id\'],\n                            self.domainB[\'id\']]),\n                       set([self.domainA[\'id\'],\n                            self.domainB[\'id\'],\n                            self.domainC[\'id\']]))\n\n        for token, domains_ref in zip(tokens, domain_refs):\n            for url in urls:\n                r = self.get(url, token=token)\n                domains_resp = r.result[\'domains\']\n                domains = set(p[\'id\'] for p in domains_resp)\n                self.assertEqual(domains_ref, domains,\n                                 \'match failed for url %s\' % url)\n\n    def test_full_workflow(self):\n        """Test \'standard\' workflow for granting access tokens.\n\n        * Issue unscoped token\n        * List available projects based on groups\n        * Scope token to one of available projects\n\n        """\n        r = self._issue_unscoped_token()\n        token_resp = render_token.render_token_response_from_model(r)[\'token\']\n        # NOTE(lbragstad): Ensure only \'saml2\' is in the method list.\n        self.assertListEqual([\'saml2\'], r.methods)\n        self.assertValidMappedUser(token_resp)\n        employee_unscoped_token_id = r.id\n        r = self.get(\'/auth/projects\', token=employee_unscoped_token_id)\n        projects = r.result[\'projects\']\n        random_project = random.randint(0, len(projects) - 1)\n        project = projects[random_project]\n\n        v3_scope_request = self._scope_request(employee_unscoped_token_id,\n                                               \'project\', project[\'id\'])\n\n        r = self.v3_create_token(v3_scope_request)\n        token_resp = r.result[\'token\']\n        self.assertIn(\'token\', token_resp[\'methods\'])\n        self.assertIn(\'saml2\', token_resp[\'methods\'])\n        self._check_project_scoped_token_attributes(token_resp, project[\'id\'])\n\n    def test_workflow_with_groups_deletion(self):\n        """Test full workflow with groups deletion before token scoping.\n\n        The test scenario is as follows:\n         - Create group ``group``\n         - Create and assign roles to ``group`` and ``project_all``\n         - Patch mapping rules for existing IdP so it issues group id\n         - Issue unscoped token with ``group``\'s id\n         - Delete group ``group``\n         - Scope token to ``project_all``\n         - Expect HTTP 500 response\n\n        """\n        # create group and role\n        group = unit.new_group_ref(domain_id=self.domainA[\'id\'])\n        group = PROVIDERS.identity_api.create_group(group)\n        role = unit.new_role_ref()\n        PROVIDERS.role_api.create_role(role[\'id\'], role)\n\n        # assign role to group and project_admins\n        PROVIDERS.assignment_api.create_grant(\n            role[\'id\'], group_id=group[\'id\'], project_id=self.project_all[\'id\']\n        )\n\n        rules = {\n            \'rules\': [\n                {\n                    \'local\': [\n                        {\n                            \'group\': {\n                                \'id\': group[\'id\']\n                            }\n                        },\n                        {\n                            \'user\': {\n                                \'name\': \'{0}\'\n                            }\n                        }\n                    ],\n                    \'remote\': [\n                        {\n                            \'type\': \'UserName\'\n                        },\n                        {\n                            \'type\': \'LastName\',\n                            \'any_one_of\': [\n                                \'Account\'\n                            ]\n                        }\n                    ]\n                }\n            ]\n        }\n\n        PROVIDERS.federation_api.update_mapping(self.mapping[\'id\'], rules)\n\n        r = self._issue_unscoped_token(assertion=\'TESTER_ASSERTION\')\n\n        # delete group\n        PROVIDERS.identity_api.delete_group(group[\'id\'])\n\n        # scope token to project_all, expect HTTP 500\n        scoped_token = self._scope_request(\n            r.id, \'project\',\n            self.project_all[\'id\'])\n\n        self.v3_create_token(\n            scoped_token, expected_status=http.client.INTERNAL_SERVER_ERROR)\n\n    def test_lists_with_missing_group_in_backend(self):\n        """Test a mapping that points to a group that does not exist.\n\n        For explicit mappings, we expect the group to exist in the backend,\n        but for lists, specifically blacklists, a missing group is expected\n        as many groups will be specified by the IdP that are not Keystone\n        groups.\n\n        The test scenario is as follows:\n         - Create group ``EXISTS``\n         - Set mapping rules for existing IdP with a blacklist\n           that passes through as REMOTE_USER_GROUPS\n         - Issue unscoped token with on group  ``EXISTS`` id in it\n\n        """\n        domain_id = self.domainA[\'id\']\n        domain_name = self.domainA[\'name\']\n        group = unit.new_group_ref(domain_id=domain_id, name=\'EXISTS\')\n        group = PROVIDERS.identity_api.create_group(group)\n        rules = {\n            \'rules\': [\n                {\n                    "local": [\n                        {\n                            "user": {\n                                "name": "{0}",\n                                "id": "{0}"\n                            }\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "REMOTE_USER"\n                        }\n                    ]\n                },\n                {\n                    "local": [\n                        {\n                            "groups": "{0}",\n                            "domain": {"name": domain_name}\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "REMOTE_USER_GROUPS",\n                        }\n                    ]\n                }\n            ]\n        }\n        PROVIDERS.federation_api.update_mapping(self.mapping[\'id\'], rules)\n        r = self._issue_unscoped_token(assertion=\'UNMATCHED_GROUP_ASSERTION\')\n        assigned_group_ids = r.federated_groups\n        self.assertEqual(1, len(assigned_group_ids))\n        self.assertEqual(group[\'id\'], assigned_group_ids[0][\'id\'])\n\n    def test_empty_blacklist_passess_all_values(self):\n        """Test a mapping with empty blacklist specified.\n\n        Not adding a ``blacklist`` keyword to the mapping rules has the same\n        effect as adding an empty ``blacklist``.\n        In both cases, the mapping engine will not discard any groups that are\n        associated with apache environment variables.\n\n        This test checks scenario where an empty blacklist was specified.\n        Expected result is to allow any value.\n\n        The test scenario is as follows:\n         - Create group ``EXISTS``\n         - Create group ``NO_EXISTS``\n         - Set mapping rules for existing IdP with a blacklist\n           that passes through as REMOTE_USER_GROUPS\n         - Issue unscoped token with groups  ``EXISTS`` and ``NO_EXISTS``\n           assigned\n\n        """\n        domain_id = self.domainA[\'id\']\n        domain_name = self.domainA[\'name\']\n\n        # Add a group "EXISTS"\n        group_exists = unit.new_group_ref(domain_id=domain_id, name=\'EXISTS\')\n        group_exists = PROVIDERS.identity_api.create_group(group_exists)\n\n        # Add a group "NO_EXISTS"\n        group_no_exists = unit.new_group_ref(domain_id=domain_id,\n                                             name=\'NO_EXISTS\')\n        group_no_exists = PROVIDERS.identity_api.create_group(group_no_exists)\n\n        group_ids = set([group_exists[\'id\'], group_no_exists[\'id\']])\n\n        rules = {\n            \'rules\': [\n                {\n                    "local": [\n                        {\n                            "user": {\n                                "name": "{0}",\n                                "id": "{0}"\n                            }\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "REMOTE_USER"\n                        }\n                    ]\n                },\n                {\n                    "local": [\n                        {\n                            "groups": "{0}",\n                            "domain": {"name": domain_name}\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "REMOTE_USER_GROUPS",\n                            "blacklist": []\n                        }\n                    ]\n                }\n            ]\n        }\n        PROVIDERS.federation_api.update_mapping(self.mapping[\'id\'], rules)\n        r = self._issue_unscoped_token(assertion=\'UNMATCHED_GROUP_ASSERTION\')\n        assigned_group_ids = r.federated_groups\n        self.assertEqual(len(group_ids), len(assigned_group_ids))\n        for group in assigned_group_ids:\n            self.assertIn(group[\'id\'], group_ids)\n\n    def test_not_adding_blacklist_passess_all_values(self):\n        """Test a mapping without blacklist specified.\n\n        Not adding a ``blacklist`` keyword to the mapping rules has the same\n        effect as adding an empty ``blacklist``. In both cases all values will\n        be accepted and passed.\n\n        This test checks scenario where an blacklist was not specified.\n        Expected result is to allow any value.\n\n        The test scenario is as follows:\n         - Create group ``EXISTS``\n         - Create group ``NO_EXISTS``\n         - Set mapping rules for existing IdP with a blacklist\n           that passes through as REMOTE_USER_GROUPS\n         - Issue unscoped token with on groups ``EXISTS`` and ``NO_EXISTS``\n           assigned\n\n        """\n        domain_id = self.domainA[\'id\']\n        domain_name = self.domainA[\'name\']\n\n        # Add a group "EXISTS"\n        group_exists = unit.new_group_ref(domain_id=domain_id,\n                                          name=\'EXISTS\')\n        group_exists = PROVIDERS.identity_api.create_group(group_exists)\n\n        # Add a group "NO_EXISTS"\n        group_no_exists = unit.new_group_ref(domain_id=domain_id,\n                                             name=\'NO_EXISTS\')\n        group_no_exists = PROVIDERS.identity_api.create_group(group_no_exists)\n\n        group_ids = set([group_exists[\'id\'], group_no_exists[\'id\']])\n\n        rules = {\n            \'rules\': [\n                {\n                    "local": [\n                        {\n                            "user": {\n                                "name": "{0}",\n                                "id": "{0}"\n                            }\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "REMOTE_USER"\n                        }\n                    ]\n                },\n                {\n    def test_invalid_scope_body(self):\n        """Test that missing the scope in request body raises an exception.\n\n        Raises exception.SchemaValidationError() - error code 400\n\n        """\n\n                        {\n                            "type": "REMOTE_USER_GROUPS",\n                        }\n                    ]\n                }\n            ]\n        }\n        PROVIDERS.federation_api.update_mapping(self.mapping[\'id\'], rules)\n    def test_invalid_token_body(self):\n        """Test that missing the token in request body raises an exception.\n\n        Raises exception.SchemaValidationError() - error code 400\n\n        """\n\n        """Test that empty whitelist blocks all the values.\n\n        Not adding a ``whitelist`` keyword to the mapping value is different\n        than adding empty whitelist.  The former case will simply pass all the\n        values, whereas the latter would discard all the values.\n\n        This test checks scenario where an empty whitelist was specified.\n        The expected result is that no groups are matched.\n\n        The test scenario is as follows:\n         - Create group ``EXISTS``\n         - Set mapping rules for existing IdP with an empty whitelist\n           that whould discard any values from the assertion\n         - Try issuing unscoped token, no groups were matched and that the\n           federated user does not have any group assigned.\n\n        """\n        domain_id = self.domainA[\'id\']\n        domain_name = self.domainA[\'name\']\n        group = unit.new_group_ref(domain_id=domain_id, name=\'EXISTS\')\n        group = PROVIDERS.identity_api.create_group(group)\n        rules = {\n            \'rules\': [\n                {\n                    "local": [\n                        {\n                            "user": {\n                                "name": "{0}",\n                                "id": "{0}"\n                            }\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "REMOTE_USER"\n                        }\n                    ]\n                },\n                {\n                    "local": [\n                        {\n                            "groups": "{0}",\n                            "domain": {"name": domain_name}\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "REMOTE_USER_GROUPS",\n                            "whitelist": []\n                        }\n                    ]\n                }\n            ]\n        }\n        PROVIDERS.federation_api.update_mapping(self.mapping[\'id\'], rules)\n        r = self._issue_unscoped_token(assertion=\'UNMATCHED_GROUP_ASSERTION\')\n        assigned_groups = r.federated_groups\n        self.assertEqual(len(assigned_groups), 0)\n\n    def test_not_setting_whitelist_accepts_all_values(self):\n        """Test that not setting whitelist passes.\n\n        Not adding a ``whitelist`` keyword to the mapping value is different\n        than adding empty whitelist.  The former case will simply pass all the\n        values, whereas the latter would discard all the values.\n\n        This test checks a scenario where a ``whitelist`` was not specified.\n        Expected result is that no groups are ignored.\n\n        The test scenario is as follows:\n         - Create group ``EXISTS``\n         - Set mapping rules for existing IdP with an empty whitelist\n           that whould discard any values from the assertion\n         - Issue an unscoped token and make sure ephemeral user is a member of\n           two groups.\n\n        """\n        domain_id = self.domainA[\'id\']\n        domain_name = self.domainA[\'name\']\n\n        # Add a group "EXISTS"\n        group_exists = unit.new_group_ref(domain_id=domain_id,\n                                          name=\'EXISTS\')\n        group_exists = PROVIDERS.identity_api.create_group(group_exists)\n\n        # Add a group "NO_EXISTS"\n        group_no_exists = unit.new_group_ref(domain_id=domain_id,\n                                             name=\'NO_EXISTS\')\n        group_no_exists = PROVIDERS.identity_api.create_group(group_no_exists)\n\n        group_ids = set([group_exists[\'id\'], group_no_exists[\'id\']])\n\n        rules = {\n            \'rules\': [\n                {\n                    "local": [\n                        {\n                            "user": {\n                                "name": "{0}",\n                                "id": "{0}"\n                            }\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "REMOTE_USER"\n                        }\n                    ]\n                },\n                {\n                    "local": [\n                        {\n                            "groups": "{0}",\n                            "domain": {"name": domain_name}\n                        }\n                    ],\n                    "remote": [\n                        {\n                            "type": "REMOTE_USER_GROUPS",\n                        }\n                    ]\n                }\n            ]\n        }\n        PROVIDERS.federation_api.update_mapping(self.mapping[\'id\'], rules)\n        r = self._issue_unscoped_token(assertion=\'UNMATCHED_GROUP_ASSERTION\')\n        assigned_group_ids = r.federated_groups\n        self.assertEqual(len(group_ids), len(assigned_group_ids))\n        for group in assigned_group_ids:\n            self.assertIn(group[\'id\'], group_ids)\n\n    def test_assertion_prefix_parameter(self):\n        """Test parameters filtering based on the prefix.\n\n        With ``assertion_prefix`` set to fixed, non default value,\n        issue an unscoped token from assertion EMPLOYEE_ASSERTION_PREFIXED.\n        Expect server to return unscoped token.\n\n        """\n        self.config_fixture.config(group=\'federation\',\n                                   assertion_prefix=self.ASSERTION_PREFIX)\n        self._issue_unscoped_token(assertion=\'EMPLOYEE_ASSERTION_PREFIXED\')\n\n    def test_assertion_prefix_parameter_expect_fail(self):\n        """Test parameters filtering based on the prefix.\n\n        With ``assertion_prefix`` default value set to empty string\n        issue an unscoped token from assertion EMPLOYEE_ASSERTION.\n        Next, configure ``assertion_prefix`` to value ``UserName``.\n        Try issuing unscoped token with EMPLOYEE_ASSERTION.\n        Expect server to raise exception.Unathorized exception.\n\n        """\n        self._issue_unscoped_token()\n        self.config_fixture.config(group=\'federation\',\n                                   assertion_prefix=\'UserName\')\n\n        self.assertRaises(exception.Unauthorized,\n                          self._issue_unscoped_token)\n\n    def test_unscoped_token_has_user_domain(self):\n        r = self._issue_unscoped_token()\n        self._check_domains_are_valid(\n            render_token.render_token_response_from_model(r)[\'token\'])\n\n    def test_scoped_token_has_user_domain(self):\n        r = self.v3_create_token(\n            self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_EMPLOYEE)\n        self._check_domains_are_valid(r.json_body[\'token\'])\n\n    def test_issue_unscoped_token_for_local_user(self):\n        r = self._issue_unscoped_token(assertion=\'LOCAL_USER_ASSERTION\')\n        self.assertListEqual([\'saml2\'], r.methods)\n        self.assertEqual(self.user[\'id\'], r.user_id)\n        self.assertEqual(self.user[\'name\'], r.user[\'name\'])\n        self.assertEqual(self.domain[\'id\'], r.user_domain[\'id\'])\n        # Make sure the token is not scoped\n        self.assertIsNone(r.domain_id)\n        self.assertIsNone(r.project_id)\n        self.assertTrue(r.unscoped)\n\n    def test_issue_token_for_local_user_user_not_found(self):\n        self.assertRaises(exception.Unauthorized,\n                          self._issue_unscoped_token,\n                          assertion=\'ANOTHER_LOCAL_USER_ASSERTION\')\n\n    def test_user_name_and_id_in_federation_token(self):\n        r = self._issue_unscoped_token(assertion=\'EMPLOYEE_ASSERTION\')\n        self.assertEqual(\n            mapping_fixtures.EMPLOYEE_ASSERTION[\'UserName\'],\n            r.user[\'name\'])\n        self.assertNotEqual(r.user[\'name\'], r.user_id)\n        r = self.v3_create_token(\n            self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_EMPLOYEE)\n        token = r.json_body[\'token\']\n        self.assertEqual(\n            mapping_fixtures.EMPLOYEE_ASSERTION[\'UserName\'],\n            token[\'user\'][\'name\'])\n        self.assertNotEqual(token[\'user\'][\'name\'], token[\'user\'][\'id\'])\n\n    def test_issue_unscoped_token_with_remote_different_from_protocol(self):\n        protocol = PROVIDERS.federation_api.get_protocol(\n            self.IDP_WITH_REMOTE, self.PROTOCOL\n        )\n        protocol[\'remote_id_attribute\'] = uuid.uuid4().hex\n        PROVIDERS.federation_api.update_protocol(\n            self.IDP_WITH_REMOTE, protocol[\'id\'], protocol\n        )\n        self._issue_unscoped_token(\n            idp=self.IDP_WITH_REMOTE,\n            environment={\n                protocol[\'remote_id_attribute\']: self.REMOTE_IDS[0]\n            }\n        )\n        self.assertRaises(\n            exception.Unauthorized,\n            self._issue_unscoped_token,\n            idp=self.IDP_WITH_REMOTE,\n            environment={uuid.uuid4().hex: self.REMOTE_IDS[0]}\n        )\n\n\nclass FernetFederatedTokenTests(test_v3.RestfulTestCase, FederatedSetupMixin):\n    AUTH_METHOD = \'token\'\n\n    def load_fixtures(self, fixtures):\n        super(FernetFederatedTokenTests, self).load_fixtures(fixtures)\n        self.load_federation_sample_data()\n\n    def config_overrides(self):\n        super(FernetFederatedTokenTests, self).config_overrides()\n        self.config_fixture.config(group=\'token\', provider=\'fernet\')\n        self.useFixture(\n            ksfixtures.KeyRepository(\n                self.config_fixture,\n                \'fernet_tokens\',\n                CONF.fernet_tokens.max_active_keys\n            )\n        )\n\n    def auth_plugin_config_override(self):\n        methods = [\'saml2\', \'token\', \'password\']\n        super(FernetFederatedTokenTests,\n              self).auth_plugin_config_override(methods)\n\n    def test_federated_unscoped_token(self):\n        resp = self._issue_unscoped_token()\n        self.assertValidMappedUser(\n            render_token.render_token_response_from_model(resp)[\'token\'])\n\n    def test_federated_unscoped_token_with_multiple_groups(self):\n        assertion = \'ANOTHER_CUSTOMER_ASSERTION\'\n        resp = self._issue_unscoped_token(assertion=assertion)\n        self.assertValidMappedUser(\n            render_token.render_token_response_from_model(resp)[\'token\'])\n\n    def test_validate_federated_unscoped_token(self):\n        resp = self._issue_unscoped_token()\n        unscoped_token = resp.id\n        # assert that the token we received is valid\n        self.get(\'/auth/tokens/\', headers={\'X-Subject-Token\': unscoped_token})\n\n    def test_fernet_full_workflow(self):\n        """Test \'standard\' workflow for granting Fernet access tokens.\n\n        * Issue unscoped token\n        * List available projects based on groups\n        * Scope token to one of available projects\n\n        """\n        resp = self._issue_unscoped_token()\n        self.assertValidMappedUser(\n            render_token.render_token_response_from_model(resp)[\'token\'])\n        unscoped_token = resp.id\n        resp = self.get(\'/auth/projects\', token=unscoped_token)\n        projects = resp.result[\'projects\']\n        random_project = random.randint(0, len(projects) - 1)\n        project = projects[random_project]\n\n        v3_scope_request = self._scope_request(unscoped_token,\n                                               \'project\', project[\'id\'])\n\n        resp = self.v3_create_token(v3_scope_request)\n        token_resp = resp.result[\'token\']\n        self._check_project_scoped_token_attributes(token_resp, project[\'id\'])\n\n\nclass JWSFederatedTokenTests(test_v3.RestfulTestCase, FederatedSetupMixin):\n    AUTH_METHOD = \'token\'\n\n    def load_fixtures(self, fixtures):\n        super(JWSFederatedTokenTests, self).load_fixtures(fixtures)\n        self.load_federation_sample_data()\n\n    def config_overrides(self):\n        super(JWSFederatedTokenTests, self).config_overrides()\n        self.config_fixture.config(group=\'token\', provider=\'jws\')\n        self.useFixture(ksfixtures.JWSKeyRepository(self.config_fixture))\n\n    def auth_plugin_config_override(self):\n        methods = [\'saml2\', \'token\', \'password\']\n        super(JWSFederatedTokenTests,\n              self).auth_plugin_config_override(methods)\n\n    def test_federated_unscoped_token(self):\n        token_model = self._issue_unscoped_token()\n        self.assertValidMappedUser(\n            render_token.render_token_response_from_model(token_model)[\'token\']\n        )\n\n    def test_federated_unscoped_token_with_multiple_groups(self):\n        assertion = \'ANOTHER_CUSTOMER_ASSERTION\'\n        token_model = self._issue_unscoped_token(assertion=assertion)\n        self.assertValidMappedUser(\n            render_token.render_token_response_from_model(token_model)[\'token\']\n        )\n\n    def test_validate_federated_unscoped_token(self):\n        token_model = self._issue_unscoped_token()\n        unscoped_token = token_model.id\n        # assert that the token we received is valid\n        self.get(\'/auth/tokens/\', headers={\'X-Subject-Token\': unscoped_token})\n\n    def test_jws_full_workflow(self):\n        """Test \'standard\' workflow for granting JWS tokens.\n\n        * Issue unscoped token\n        * List available projects based on groups\n        * Scope token to one of available projects\n\n        """\n        token_model = self._issue_unscoped_token()\n        self.assertValidMappedUser(\n            render_token.render_token_response_from_model(token_model)[\'token\']\n        )\n        unscoped_token = token_model.id\n        resp = self.get(\'/auth/projects\', token=unscoped_token)\n        projects = resp.result[\'projects\']\n        random_project = random.randint(0, len(projects) - 1)\n        project = projects[random_project]\n\n        v3_scope_request = self._scope_request(unscoped_token,\n                                               \'project\', project[\'id\'])\n\n        resp = self.v3_create_token(v3_scope_request)\n        token_resp = resp.result[\'token\']\n        self._check_project_scoped_token_attributes(token_resp, project[\'id\'])\n\n\nclass FederatedTokenTestsMethodToken(FederatedTokenTests):\n    """Test federation operation with unified scoping auth method.\n\n    Test all the operations with auth method set to ``token`` as a new, unified\n    way for scoping all the tokens.\n\n    """\n\n    AUTH_METHOD = \'token\'\n\n    def auth_plugin_config_override(self):\n        methods = [\'saml2\', \'token\']\n        super(FederatedTokenTests,\n              self).auth_plugin_config_override(methods)\n\n    def test_full_workflow(self):\n        """Test \'standard\' workflow for granting access tokens.\n\n        * Issue unscoped token\n        * List available projects based on groups\n        * Scope token to one of available projects\n\n        """\n        r = self._issue_unscoped_token()\n        token_resp = render_token.render_token_response_from_model(r)[\'token\']\n        # NOTE(lbragstad): Ensure only \'saml2\' is in the method list.\n        self.assertListEqual([\'saml2\'], r.methods)\n        self.assertValidMappedUser(token_resp)\n        employee_unscoped_token_id = r.id\n        r = self.get(\'/auth/projects\', token=employee_unscoped_token_id)\n        projects = r.result[\'projects\']\n        random_project = random.randint(0, len(projects) - 1)\n        project = projects[random_project]\n\n        v3_scope_request = self._scope_request(employee_unscoped_token_id,\n                                               \'project\', project[\'id\'])\n\n        r = self.v3_create_token(v3_scope_request)\n        token_resp = r.result[\'token\']\n        self.assertIn(\'token\', token_resp[\'methods\'])\n        self.assertIn(\'saml2\', token_resp[\'methods\'])\n        self._check_project_scoped_token_attributes(token_resp, project[\'id\'])\n\n\nclass FederatedUserTests(test_v3.RestfulTestCase, FederatedSetupMixin):\n    """Test for federated users.\n\n    Tests new shadow users functionality\n\n    """\n\n    def auth_plugin_config_override(self):\n        methods = [\'saml2\', \'token\']\n        super(FederatedUserTests, self).auth_plugin_config_override(methods)\n\n    def load_fixtures(self, fixtures):\n        super(FederatedUserTests, self).load_fixtures(fixtures)\n        self.load_federation_sample_data()\n\n    def test_user_id_persistense(self):\n        """Ensure user_id is persistend for multiple federated authn calls."""\n        r = self._issue_unscoped_token()\n        user_id = r.user_id\n        self.assertNotEmpty(PROVIDERS.identity_api.get_user(user_id))\n\n        r = self._issue_unscoped_token()\n        user_id2 = r.user_id\n        self.assertNotEmpty(PROVIDERS.identity_api.get_user(user_id2))\n        self.assertEqual(user_id, user_id2)\n\n    def test_user_role_assignment(self):\n        # create project and role\n        project_ref = unit.new_project_ref(\n            domain_id=CONF.identity.default_domain_id)\n        PROVIDERS.resource_api.create_project(project_ref[\'id\'], project_ref)\n        role_ref = unit.new_role_ref()\n        PROVIDERS.role_api.create_role(role_ref[\'id\'], role_ref)\n\n        # authenticate via saml get back a user id\n        user_id, unscoped_token = self._authenticate_via_saml()\n\n        # exchange an unscoped token for a scoped token; resulting in\n        # unauthorized because the user doesn\'t have any role assignments\n        v3_scope_request = self._scope_request(unscoped_token, \'project\',\n                                               project_ref[\'id\'])\n        r = self.v3_create_token(v3_scope_request,\n                                 expected_status=http.client.UNAUTHORIZED)\n\n        # assign project role to federated user\n        PROVIDERS.assignment_api.add_role_to_user_and_project(\n            user_id, project_ref[\'id\'], role_ref[\'id\'])\n\n        # exchange an unscoped token for a scoped token\n        r = self.v3_create_token(v3_scope_request,\n                                 expected_status=http.client.CREATED)\n        scoped_token = r.headers[\'X-Subject-Token\']\n\n        # ensure user can access resource based on role assignment\n        path = \'/projects/%(project_id)s\' % {\'project_id\': project_ref[\'id\']}\n        r = self.v3_request(path=path, method=\'GET\',\n                            expected_status=http.client.OK,\n                            token=scoped_token)\n        self.assertValidProjectResponse(r, project_ref)\n\n        # create a 2nd project\n        project_ref2 = unit.new_project_ref(\n            domain_id=CONF.identity.default_domain_id)\n        PROVIDERS.resource_api.create_project(project_ref2[\'id\'], project_ref2)\n\n        # ensure the user cannot access the 2nd resource (forbidden)\n        path = \'/projects/%(project_id)s\' % {\'project_id\': project_ref2[\'id\']}\n        r = self.v3_request(path=path, method=\'GET\',\n                            expected_status=http.client.FORBIDDEN,\n                            token=scoped_token)\n\n    def test_domain_scoped_user_role_assignment(self):\n        # create domain and role\n        domain_ref = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(domain_ref[\'id\'], domain_ref)\n        role_ref = unit.new_role_ref()\n        PROVIDERS.role_api.create_role(role_ref[\'id\'], role_ref)\n\n        # authenticate via saml get back a user id\n        user_id, unscoped_token = self._authenticate_via_saml()\n\n        # exchange an unscoped token for a scoped token; resulting in\n        # unauthorized because the user doesn\'t have any role assignments\n        v3_scope_request = self._scope_request(unscoped_token, \'domain\',\n                                               domain_ref[\'id\'])\n        r = self.v3_create_token(v3_scope_request,\n                                 expected_status=http.client.UNAUTHORIZED)\n\n        # assign domain role to user\n        PROVIDERS.assignment_api.create_grant(\n            role_ref[\'id\'], user_id=user_id, domain_id=domain_ref[\'id\']\n        )\n\n        # exchange an unscoped token for domain scoped token and test\n        r = self.v3_create_token(v3_scope_request,\n                                 expected_status=http.client.CREATED)\n        self.assertIsNotNone(r.headers.get(\'X-Subject-Token\'))\n        token_resp = r.result[\'token\']\n        self.assertIn(\'domain\', token_resp)\n\n    def test_auth_projects_matches_federation_projects(self):\n        # create project and role\n        project_ref = unit.new_project_ref(\n            domain_id=CONF.identity.default_domain_id)\n        PROVIDERS.resource_api.create_project(project_ref[\'id\'], project_ref)\n        role_ref = unit.new_role_ref()\n        PROVIDERS.role_api.create_role(role_ref[\'id\'], role_ref)\n\n        # authenticate via saml get back a user id\n        user_id, unscoped_token = self._authenticate_via_saml()\n\n        # assign project role to federated user\n        PROVIDERS.assignment_api.add_role_to_user_and_project(\n            user_id, project_ref[\'id\'], role_ref[\'id\'])\n\n        # get auth projects\n        r = self.get(\'/auth/projects\', token=unscoped_token)\n        auth_projects = r.result[\'projects\']\n\n        # get federation projects\n        r = self.get(\'/OS-FEDERATION/projects\', token=unscoped_token)\n        fed_projects = r.result[\'projects\']\n\n        # compare\n        self.assertCountEqual(auth_projects, fed_projects)\n\n    def test_auth_projects_matches_federation_projects_with_group_assign(self):\n        # create project, role, group\n        domain_id = CONF.identity.default_domain_id\n        project_ref = unit.new_project_ref(domain_id=domain_id)\n        PROVIDERS.resource_api.create_project(project_ref[\'id\'], project_ref)\n        role_ref = unit.new_role_ref()\n        PROVIDERS.role_api.create_role(role_ref[\'id\'], role_ref)\n        group_ref = unit.new_group_ref(domain_id=domain_id)\n        group_ref = PROVIDERS.identity_api.create_group(group_ref)\n\n        # authenticate via saml get back a user id\n        user_id, unscoped_token = self._authenticate_via_saml()\n\n        # assign role to group at project\n        PROVIDERS.assignment_api.create_grant(\n            role_ref[\'id\'], group_id=group_ref[\'id\'],\n            project_id=project_ref[\'id\'], domain_id=domain_id\n        )\n\n        # add user to group\n        PROVIDERS.identity_api.add_user_to_group(\n            user_id=user_id, group_id=group_ref[\'id\']\n        )\n\n        # get auth projects\n        r = self.get(\'/auth/projects\', token=unscoped_token)\n        auth_projects = r.result[\'projects\']\n\n        # get federation projects\n        r = self.get(\'/OS-FEDERATION/projects\', token=unscoped_token)\n        fed_projects = r.result[\'projects\']\n\n        # compare\n        self.assertCountEqual(auth_projects, fed_projects)\n\n    def test_auth_domains_matches_federation_domains(self):\n        # create domain and role\n        domain_ref = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(domain_ref[\'id\'], domain_ref)\n        role_ref = unit.new_role_ref()\n        PROVIDERS.role_api.create_role(role_ref[\'id\'], role_ref)\n\n        # authenticate via saml get back a user id and token\n        user_id, unscoped_token = self._authenticate_via_saml()\n\n        # assign domain role to user\n        PROVIDERS.assignment_api.create_grant(\n            role_ref[\'id\'], user_id=user_id, domain_id=domain_ref[\'id\']\n        )\n\n        # get auth domains\n        r = self.get(\'/auth/domains\', token=unscoped_token)\n        auth_domains = r.result[\'domains\']\n\n        # get federation domains\n        r = self.get(\'/OS-FEDERATION/domains\', token=unscoped_token)\n        fed_domains = r.result[\'domains\']\n\n        # compare\n        self.assertCountEqual(auth_domains, fed_domains)\n\n    def test_auth_domains_matches_federation_domains_with_group_assign(self):\n        # create role, group, and domain\n        domain_ref = unit.new_domain_ref()\n        PROVIDERS.resource_api.create_domain(domain_ref[\'id\'], domain_ref)\n        role_ref = unit.new_role_ref()\n        PROVIDERS.role_api.create_role(role_ref[\'id\'], role_ref)\n        group_ref = unit.new_group_ref(domain_id=domain_ref[\'id\'])\n        group_ref = PROVIDERS.identity_api.create_group(group_ref)\n\n        # authenticate via saml get back a user id and token\n        user_id, unscoped_token = self._authenticate_via_saml()\n\n        # assign domain role to group\n        PROVIDERS.assignment_api.create_grant(\n            role_ref[\'id\'], group_id=group_ref[\'id\'],\n            domain_id=domain_ref[\'id\']\n        )\n\n        # add user to group\n        PROVIDERS.identity_api.add_user_to_group(\n            user_id=user_id, group_id=group_ref[\'id\']\n        )\n\n        # get auth domains\n        r = self.get(\'/auth/domains\', token=unscoped_token)\n        auth_domains = r.result[\'domains\']\n\n        # get federation domains\n        r = self.get(\'/OS-FEDERATION/domains\', token=unscoped_token)\n        fed_domains = r.result[\'domains\']\n\n        # compare\n        self.assertCountEqual(auth_domains, fed_domains)\n\n    def test_list_head_domains_for_user_duplicates(self):\n        # create role\n        role_ref = unit.new_role_ref()\n        PROVIDERS.role_api.create_role(role_ref[\'id\'], role_ref)\n\n        # authenticate via saml get back a user id and token\n        user_id, unscoped_token = self._authenticate_via_saml()\n\n        # get federation group domains\n        r = self.get(\'/OS-FEDERATION/domains\', token=unscoped_token)\n        group_domains = r.result[\'domains\']\n        domain_from_group = group_domains[0]\n\n        self.head(\n            \'/OS-FEDERATION/domains\',\n            token=unscoped_token,\n            expected_status=http.client.OK\n        )\n\n        # assign group domain and role to user, this should create a\n        # duplicate domain\n        PROVIDERS.assignment_api.create_grant(\n            role_ref[\'id\'], user_id=user_id, domain_id=domain_from_group[\'id\']\n        )\n\n        # get user domains via /OS-FEDERATION/domains and test for duplicates\n        r = self.get(\'/OS-FEDERATION/domains\', token=unscoped_token)\n        user_domains = r.result[\'domains\']\n        user_domain_ids = []\n        for domain in user_domains:\n            self.assertNotIn(domain[\'id\'], user_domain_ids)\n            user_domain_ids.append(domain[\'id\'])\n\n        # get user domains via /auth/domains and test for duplicates\n        r = self.get(\'/auth/domains\', token=unscoped_token)\n        user_domains = r.result[\'domains\']\n        user_domain_ids = []\n        for domain in user_domains:\n            self.assertNotIn(domain[\'id\'], user_domain_ids)\n            user_domain_ids.append(domain[\'id\'])\n\n    def test_list_head_projects_for_user_duplicates(self):\n        # create role\n        role_ref = unit.new_role_ref()\n        PROVIDERS.role_api.create_role(role_ref[\'id\'], role_ref)\n\n        # authenticate via saml get back a user id and token\n        user_id, unscoped_token = self._authenticate_via_saml()\n\n        # get federation group projects\n        r = self.get(\'/OS-FEDERATION/projects\', token=unscoped_token)\n        group_projects = r.result[\'projects\']\n        project_from_group = group_projects[0]\n\n        self.head(\n            \'/OS-FEDERATION/projects\',\n            token=unscoped_token,\n            expected_status=http.client.OK\n        )\n\n        # assign group project and role to user, this should create a\n        # duplicate project\n        PROVIDERS.assignment_api.add_role_to_user_and_project(\n            user_id, project_from_group[\'id\'], role_ref[\'id\'])\n\n        # get user projects via /OS-FEDERATION/projects and test for duplicates\n        r = self.get(\'/OS-FEDERATION/projects\', token=unscoped_token)\n        user_projects = r.result[\'projects\']\n        user_project_ids = []\n        for project in user_projects:\n            self.assertNotIn(project[\'id\'], user_project_ids)\n            user_project_ids.append(project[\'id\'])\n\n        # get user projects via /auth/projects and test for duplicates\n        r = self.get(\'/auth/projects\', token=unscoped_token)\n        user_projects = r.result[\'projects\']\n        user_project_ids = []\n        for project in user_projects:\n            self.assertNotIn(project[\'id\'], user_project_ids)\n            user_project_ids.append(project[\'id\'])\n\n    def test_delete_protocol_after_federated_authentication(self):\n        # Create a protocol\n        protocol = self.proto_ref(mapping_id=self.mapping[\'id\'])\n        PROVIDERS.federation_api.create_protocol(\n            self.IDP, protocol[\'id\'], protocol)\n\n        # Authenticate to create a new federated_user entry with a foreign\n        # key pointing to the protocol\n        r = self._issue_unscoped_token()\n        user_id = r.user_id\n        self.assertNotEmpty(PROVIDERS.identity_api.get_user(user_id))\n\n        # Now we should be able to delete the protocol\n        PROVIDERS.federation_api.delete_protocol(self.IDP, protocol[\'id\'])\n\n    def _authenticate_via_saml(self):\n        r = self._issue_unscoped_token()\n        unscoped_token = r.id\n        token_resp = render_token.render_token_response_from_model(r)[\'token\']\n        self.assertValidMappedUser(token_resp)\n        return r.user_id, unscoped_token\n\n\nclass ShadowMappingTests(test_v3.RestfulTestCase, FederatedSetupMixin):\n    """Test class dedicated to auto-provisioning resources at login.\n\n    A shadow mapping is a mapping that contains extra properties about that\n    specific federated user\'s situation based on attributes from the assertion.\n    For example, a shadow mapping can tell us that a user should have specific\n    role assignments on certain projects within a domain. When a federated user\n    authenticates, the shadow mapping will create these entities before\n    returning the authenticated response to the user. This test class is\n    dedicated to testing specific aspects of shadow mapping when performing\n    federated authentication.\n    """\n\n    def setUp(self):\n        super(ShadowMappingTests, self).setUp()\n        # update the mapping we have already setup to have specific projects\n        # and roles.\n        PROVIDERS.federation_api.update_mapping(\n            self.mapping[\'id\'],\n            mapping_fixtures.MAPPING_PROJECTS\n        )\n\n        # The shadow mapping we\'re using in these tests contain a role named\n        # `member` and `observer` for the sake of using something other than\n        # `admin`. We\'ll need to create those before hand, otherwise the\n        # mapping will fail during authentication because the roles defined in\n        # the mapping do not exist yet. The shadow mapping mechanism currently\n        # doesn\'t support creating roles on-the-fly, but this could change in\n        # the future after we get some feedback from shadow mapping being used\n        # in real deployments. We also want to make sure we are dealing with\n        # global roles and not domain-scoped roles. We have specific tests\n        # below that test that behavior and the setup is done in the test.\n        member_role_ref = unit.new_role_ref(name=\'member\')\n        assert member_role_ref[\'domain_id\'] is None\n        self.member_role = PROVIDERS.role_api.create_role(\n            member_role_ref[\'id\'], member_role_ref\n        )\n        observer_role_ref = unit.new_role_ref(name=\'observer\')\n        assert observer_role_ref[\'domain_id\'] is None\n        self.observer_role = PROVIDERS.role_api.create_role(\n            observer_role_ref[\'id\'], observer_role_ref\n        )\n\n        # This is a mapping of the project name to the role that is supposed to\n        # be assigned to the user on that project from the shadow mapping.\n        self.expected_results = {\n            \'Production\': \'observer\',\n            \'Staging\': \'member\',\n            \'Project for tbo\': \'admin\'\n        }\n\n    def auth_plugin_config_override(self):\n        methods = [\'saml2\', \'token\']\n        super(ShadowMappingTests, self).auth_plugin_config_override(methods)\n\n    def load_fixtures(self, fixtures):\n        super(ShadowMappingTests, self).load_fixtures(fixtures)\n        self.load_federation_sample_data()\n\n    def test_shadow_mapping_creates_projects(self):\n        projects = PROVIDERS.resource_api.list_projects()\n        for project in projects:\n            self.assertNotIn(project[\'name\'], self.expected_results)\n\n        response = self._issue_unscoped_token()\n        self.assertValidMappedUser(\n            render_token.render_token_response_from_model(response)[\'token\'])\n        unscoped_token = response.id\n        response = self.get(\'/auth/projects\', token=unscoped_token)\n        projects = response.json_body[\'projects\']\n        for project in projects:\n            project = PROVIDERS.resource_api.get_project_by_name(\n                project[\'name\'],\n                self.idp[\'domain_id\']\n            )\n            self.assertIn(project[\'name\'], self.expected_results)\n\n    def test_shadow_mapping_create_projects_role_assignments(self):\n        response = self._issue_unscoped_token()\n        self.assertValidMappedUser(\n            render_token.render_token_response_from_model(response)[\'token\'])\n        unscoped_token = response.id\n        response = self.get(\'/auth/projects\', token=unscoped_token)\n        projects = response.json_body[\'projects\']\n        for project in projects:\n            # Ask for a scope token to each project in the mapping. Each token\n            # should contain a different role so let\'s check that is right,\n            # too.\n            scope = self._scope_request(\n                unscoped_token, \'project\', project[\'id\']\n            )\n            response = self.v3_create_token(scope)\n            project_name = response.json_body[\'token\'][\'project\'][\'name\']\n            roles = response.json_body[\'token\'][\'roles\']\n            self.assertEqual(\n                self.expected_results[project_name], roles[0][\'name\']\n            )\n\n    def test_shadow_mapping_does_not_create_roles(self):\n        # If a role required by the mapping does not exist, then we should fail\n        # the mapping since shadow mapping currently does not support creating\n        # mappings on-the-fly.\n        PROVIDERS.role_api.delete_role(self.observer_role[\'id\'])\n        self.assertRaises(exception.RoleNotFound, self._issue_unscoped_token)\n\n    def test_shadow_mapping_creates_project_in_identity_provider_domain(self):\n        response = self._issue_unscoped_token()\n        self.assertValidMappedUser(\n            render_token.render_token_response_from_model(response)[\'token\'])\n        unscoped_token = response.id\n        response = self.get(\'/auth/projects\', token=unscoped_token)\n        projects = response.json_body[\'projects\']\n        for project in projects:\n            self.assertEqual(project[\'domain_id\'], self.idp[\'domain_id\'])\n\n    def test_shadow_mapping_is_idempotent(self):\n        """Test that projects remain idempotent for every federated auth."""\n        response = self._issue_unscoped_token()\n        self.assertValidMappedUser(\n            render_token.render_token_response_from_model(response)[\'token\'])\n        unscoped_token = response.id\n        response = self.get(\'/auth/projects\', token=unscoped_token)\n        project_ids = [p[\'id\'] for p in response.json_body[\'projects\']]\n        response = self._issue_unscoped_token()\n        unscoped_token = response.id\n        response = self.get(\'/auth/projects\', token=unscoped_token)\n        projects = response.json_body[\'projects\']\n        for project in projects:\n            self.assertIn(project[\'id\'], project_ids)\n\n    def test_roles_outside_idp_domain_fail_mapping(self):\n        # Create a new domain\n        d = unit.new_domain_ref()\n        new_domain = PROVIDERS.resource_api.create_domain(d[\'id\'], d)\n\n        # Delete the member role and recreate it in a different domain\n        PROVIDERS.role_api.delete_role(self.member_role[\'id\'])\n        member_role_ref = unit.new_role_ref(\n            name=\'member\',\n            domain_id=new_domain[\'id\']\n        )\n        PROVIDERS.role_api.create_role(member_role_ref[\'id\'], member_role_ref)\n        self.assertRaises(\n            exception.DomainSpecificRoleNotWithinIdPDomain,\n            self._issue_unscoped_token\n        )\n\n    def test_roles_in_idp_domain_can_be_assigned_from_mapping(self):\n        # Delete the member role and recreate it in the domain of the idp\n        PROVIDERS.role_api.delete_role(self.member_role[\'id\'])\n        member_role_ref = unit.new_role_ref(\n            name=\'member\',\n            domain_id=self.idp[\'domain_id\']\n        )\n        PROVIDERS.role_api.create_role(member_role_ref[\'id\'], member_role_ref)\n        response = self._issue_unscoped_token()\n        user_id = response.user_id\n        unscoped_token = response.id\n        response = self.get(\'/auth/projects\', token=unscoped_token)\n        projects = response.json_body[\'projects\']\n        staging_project = PROVIDERS.resource_api.get_project_by_name(\n            \'Staging\', self.idp[\'domain_id\']\n        )\n        for project in projects:\n            # Even though the mapping successfully assigned the Staging project\n            # a member role for our user, the /auth/projects response doesn\'t\n            # include projects with only domain-specific role assignments.\n            self.assertNotEqual(project[\'name\'], \'Staging\')\n        domain_role_assignments = (\n            PROVIDERS.assignment_api.list_role_assignments(\n                user_id=user_id,\n                project_id=staging_project[\'id\'],\n                strip_domain_roles=False\n            )\n        )\n        self.assertEqual(\n            staging_project[\'id\'], domain_role_assignments[0][\'project_id\']\n        )\n        self.assertEqual(\n            user_id, domain_role_assignments[0][\'user_id\']\n        )\n\n    def test_mapping_with_groups_includes_projects_with_group_assignment(self):\n        # create a group called Observers\n        observer_group = unit.new_group_ref(\n            domain_id=self.idp[\'domain_id\'],\n            name=\'Observers\'\n        )\n        observer_group = PROVIDERS.identity_api.create_group(observer_group)\n        # make sure the Observers group has a role on the finance project\n        finance_project = unit.new_project_ref(\n            domain_id=self.idp[\'domain_id\'],\n            name=\'Finance\'\n        )\n        finance_project = PROVIDERS.resource_api.create_project(\n            finance_project[\'id\'], finance_project\n        )\n        PROVIDERS.assignment_api.create_grant(\n            self.observer_role[\'id\'],\n            group_id=observer_group[\'id\'],\n            project_id=finance_project[\'id\']\n        )\n        # update the mapping\n        group_rule = {\n            \'group\': {\n                \'name\': \'Observers\',\n                \'domain\': {\n                    \'id\': self.idp[\'domain_id\']\n                }\n            }\n        }\n        updated_mapping = copy.deepcopy(mapping_fixtures.MAPPING_PROJECTS)\n        updated_mapping[\'rules\'][0][\'local\'].append(group_rule)\n        PROVIDERS.federation_api.update_mapping(\n            self.mapping[\'id\'], updated_mapping\n        )\n        response = self._issue_unscoped_token()\n        # user_id = response.json_body[\'token\'][\'user\'][\'id\']\n        unscoped_token = response.id\n        response = self.get(\'/auth/projects\', token=unscoped_token)\n        projects = response.json_body[\'projects\']\n        self.expected_results = {\n            # These assignments are all a result of a direct mapping from the\n            # shadow user to the newly created project.\n            \'Production\': \'observer\',\n            \'Staging\': \'member\',\n            \'Project for tbo\': \'admin\',\n            # This is a result of the mapping engine maintaining its old\n            # behavior.\n            \'Finance\': \'observer\'\n        }\n        for project in projects:\n            # Ask for a scope token to each project in the mapping. Each token\n            # should contain a different role so let\'s check that is right,\n            # too.\n            scope = self._scope_request(\n                unscoped_token, \'project\', project[\'id\']\n            )\n            response = self.v3_create_token(scope)\n            project_name = response.json_body[\'token\'][\'project\'][\'name\']\n            roles = response.json_body[\'token\'][\'roles\']\n            self.assertEqual(\n                self.expected_results[project_name], roles[0][\'name\']\n            )\n\n    def test_user_gets_only_assigned_roles(self):\n        # in bug 1677723 user could get roles outside of what was assigned\n        # to them. This test verifies that this is no longer true.\n        # Authenticate once to create the projects\n        response = self._issue_unscoped_token()\n        self.assertValidMappedUser(\n            render_token.render_token_response_from_model(response)[\'token\'])\n\n        # Assign admin role to newly-created project to another user\n        staging_project = PROVIDERS.resource_api.get_project_by_name(\n            \'Staging\', self.idp[\'domain_id\']\n        )\n        admin = unit.new_user_ref(CONF.identity.default_domain_id)\n        PROVIDERS.identity_api.create_user(admin)\n        PROVIDERS.assignment_api.create_grant(\n            self.role_admin[\'id\'], user_id=admin[\'id\'],\n            project_id=staging_project[\'id\']\n        )\n\n        # Authenticate again with the federated user and verify roles\n        response = self._issue_unscoped_token()\n        self.assertValidMappedUser(\n            render_token.render_token_response_from_model(response)[\'token\'])\n        unscoped_token = response.id\n        scope = self._scope_request(\n            unscoped_token, \'project\', staging_project[\'id\']\n        )\n        response = self.v3_create_token(scope)\n        roles = response.json_body[\'token\'][\'roles\']\n        role_ids = [r[\'id\'] for r in roles]\n        self.assertNotIn(self.role_admin[\'id\'], role_ids)\n\n\nclass JsonHomeTests(test_v3.RestfulTestCase, test_v3.JsonHomeTestMixin):\n    JSON_HOME_DATA = {\n        \'https://docs.openstack.org/api/openstack-identity/3/ext/OS-FEDERATION\'\n        \'/1.0/rel/identity_provider\': {\n            \'href-template\': \'/OS-FEDERATION/identity_providers/{idp_id}\',\n            \'href-vars\': {\n                \'idp_id\': \'https://docs.openstack.org/api/openstack-identity/3\'\n                \'/ext/OS-FEDERATION/1.0/param/idp_id\'\n            },\n        },\n    }\n\n\ndef _is_xmlsec1_installed():\n    p = subprocess.Popen(\n        [\'which\', \'xmlsec1\'],\n        stdout=subprocess.PIPE,\n        stderr=subprocess.PIPE)\n\n    # invert the return code\n    return not bool(p.wait())\n\n\ndef _load_xml(filename):\n    with open(os.path.join(XMLDIR, filename), \'r\') as xml:\n        return xml.read()\n\n\nclass SAMLGenerationTests(test_v3.RestfulTestCase):\n\n    SP_AUTH_URL = (\'http://beta.com:5000/v3/OS-FEDERATION/identity_providers\'\n                   \'/BETA/protocols/saml2/auth\')\n\n    ASSERTION_FILE = \'signed_saml2_assertion.xml\'\n\n    # The values of the following variables match the attributes values found\n    # in ASSERTION_FILE\n    ISSUER = \'https://acme.com/FIM/sps/openstack/saml20\'\n    RECIPIENT = \'http://beta.com/Shibboleth.sso/SAML2/POST\'\n    SUBJECT = \'test_user\'\n    SUBJECT_DOMAIN = \'user_domain\'\n    ROLES = [\'admin\', \'member\']\n    PROJECT = \'development\'\n    PROJECT_DOMAIN = \'project_domain\'\n    GROUPS = [\'JSON:{"name":"group1","domain":{"name":"Default"}}\',\n              \'JSON:{"name":"group2","domain":{"name":"Default"}}\']\n    SAML_GENERATION_ROUTE = \'/auth/OS-FEDERATION/saml2\'\n    ECP_GENERATION_ROUTE = \'/auth/OS-FEDERATION/saml2/ecp\'\n    ASSERTION_VERSION = "2.0"\n    SERVICE_PROVDIER_ID = \'ACME\'\n\n    def setUp(self):\n        super(SAMLGenerationTests, self).setUp()\n        self.signed_assertion = saml2.create_class_from_xml_string(\n            saml.Assertion, _load_xml(self.ASSERTION_FILE))\n        self.sp = core.new_service_provider_ref(\n            auth_url=self.SP_AUTH_URL, sp_url=self.RECIPIENT\n        )\n        url = \'/OS-FEDERATION/service_providers/\' + self.SERVICE_PROVDIER_ID\n        self.put(url, body={\'service_provider\': self.sp},\n                 expected_status=http.client.CREATED)\n\n    def test_samlize_token_values(self):\n        """Test the SAML generator produces a SAML object.\n\n        Test the SAML generator directly by passing known arguments, the result\n        should be a SAML object that consistently includes attributes based on\n        the known arguments that were passed in.\n\n        """\n        with mock.patch.object(keystone_idp, \'_sign_assertion\',\n                               return_value=self.signed_assertion):\n            generator = keystone_idp.SAMLGenerator()\n            response = generator.samlize_token(self.ISSUER, self.RECIPIENT,\n                                               self.SUBJECT,\n                                               self.SUBJECT_DOMAIN,\n                                               self.ROLES, self.PROJECT,\n                                               self.PROJECT_DOMAIN,\n                                               self.GROUPS)\n\n        assertion = response.assertion\n        self.assertIsNotNone(assertion)\n        self.assertIsInstance(assertion, saml.Assertion)\n        issuer = response.issuer\n        self.assertEqual(self.RECIPIENT, response.destination)\n        self.assertEqual(self.ISSUER, issuer.text)\n\n        user_attribute = assertion.attribute_statement[0].attribute[0]\n        self.assertEqual(self.SUBJECT, user_attribute.attribute_value[0].text)\n\n        user_domain_attribute = (\n            assertion.attribute_statement[0].attribute[1])\n        self.assertEqual(self.SUBJECT_DOMAIN,\n                         user_domain_attribute.attribute_value[0].text)\n\n        role_attribute = assertion.attribute_statement[0].attribute[2]\n        for attribute_value in role_attribute.attribute_value:\n            self.assertIn(attribute_value.text, self.ROLES)\n\n        project_attribute = assertion.attribute_statement[0].attribute[3]\n        self.assertEqual(self.PROJECT,\n                         project_attribute.attribute_value[0].text)\n\n        project_domain_attribute = (\n            assertion.attribute_statement[0].attribute[4])\n        self.assertEqual(self.PROJECT_DOMAIN,\n                         project_domain_attribute.attribute_value[0].text)\n\n        group_attribute = assertion.attribute_statement[0].attribute[5]\n        for attribute_value in group_attribute.attribute_value:\n            self.assertIn(attribute_value.text, self.GROUPS)\n\n    def test_comma_in_certfile_path(self):\n        self.config_fixture.config(\n            group=\'saml\',\n            certfile=CONF.saml.certfile + \',\')\n        generator = keystone_idp.SAMLGenerator()\n        self.assertRaises(\n            exception.UnexpectedError,\n            generator.samlize_token,\n            self.ISSUER,\n            self.RECIPIENT,\n            self.SUBJECT,\n            self.SUBJECT_DOMAIN,\n            self.ROLES,\n            self.PROJECT,\n            self.PROJECT_DOMAIN,\n            self.GROUPS)\n\n    def test_comma_in_keyfile_path(self):\n        self.config_fixture.config(\n            group=\'saml\',\n            keyfile=CONF.saml.keyfile + \',\')\n        generator = keystone_idp.SAMLGenerator()\n        self.assertRaises(\n            exception.UnexpectedError,\n            generator.samlize_token,\n            self.ISSUER,\n            self.RECIPIENT,\n            self.SUBJECT,\n            self.SUBJECT_DOMAIN,\n            self.ROLES,\n            self.PROJECT,\n            self.PROJECT_DOMAIN,\n            self.GROUPS)\n\n    def test_verify_assertion_object(self):\n        """Test that the Assertion object is built properly.\n\n        The Assertion doesn\'t need to be signed in this test, so\n        _sign_assertion method is patched and doesn\'t alter the assertion.\n\n        """\n        with mock.patch.object(keystone_idp, \'_sign_assertion\',\n                               side_effect=lambda x: x):\n            generator = keystone_idp.SAMLGenerator()\n            response = generator.samlize_token(self.ISSUER, self.RECIPIENT,\n                                               self.SUBJECT,\n                                               self.SUBJECT_DOMAIN,\n                                               self.ROLES, self.PROJECT,\n                                               self.PROJECT_DOMAIN,\n                                               self.GROUPS)\n        assertion = response.assertion\n        self.assertEqual(self.ASSERTION_VERSION, assertion.version)\n\n    def test_valid_saml_xml(self):\n        """Test the generated SAML object can become valid XML.\n\n        Test the generator directly by passing known arguments, the result\n        should be a SAML object that consistently includes attributes based on\n        the known arguments that were passed in.\n\n        """\n        with mock.patch.object(keystone_idp, \'_sign_assertion\',\n                               return_value=self.signed_assertion):\n            generator = keystone_idp.SAMLGenerator()\n            response = generator.samlize_token(self.ISSUER, self.RECIPIENT,\n                                               self.SUBJECT,\n                                               self.SUBJECT_DOMAIN,\n                                               self.ROLES, self.PROJECT,\n                                               self.PROJECT_DOMAIN,\n                                               self.GROUPS)\n\n        saml_str = response.to_string()\n        response = etree.fromstring(saml_str)\n        issuer = response[0]\n        assertion = response[2]\n\n        self.assertEqual(self.RECIPIENT, response.get(\'Destination\'))\n        self.assertEqual(self.ISSUER, issuer.text)\n\n        user_attribute = assertion[4][0]\n        self.assertEqual(self.SUBJECT, user_attribute[0].text)\n\n        user_domain_attribute = assertion[4][1]\n        self.assertEqual(self.SUBJECT_DOMAIN, user_domain_attribute[0].text)\n\n        role_attribute = assertion[4][2]\n        for attribute_value in role_attribute:\n            self.assertIn(attribute_value.text, self.ROLES)\n\n        project_attribute = assertion[4][3]\n        self.assertEqual(self.PROJECT, project_attribute[0].text)\n\n        project_domain_attribute = assertion[4][4]\n        self.assertEqual(self.PROJECT_DOMAIN, project_domain_attribute[0].text)\n\n        group_attribute = assertion[4][5]\n        for attribute_value in group_attribute:\n            self.assertIn(attribute_value.text, self.GROUPS)\n\n    def test_assertion_using_explicit_namespace_prefixes(self):\n        def mocked_subprocess_check_output(*popenargs, **kwargs):\n            # the last option is the assertion file to be signed\n            if popenargs[0] != [\'/usr/bin/which\', CONF.saml.xmlsec1_binary]:\n                filename = popenargs[0][-1]\n                with open(filename, \'r\') as f:\n                    assertion_content = f.read()\n                # since we are not testing the signature itself, we can return\n                # the assertion as is without signing it\n                return assertion_content\n\n        with mock.patch.object(subprocess, \'check_output\',\n                               side_effect=mocked_subprocess_check_output):\n            generator = keystone_idp.SAMLGenerator()\n            response = generator.samlize_token(self.ISSUER, self.RECIPIENT,\n                                               self.SUBJECT,\n                                               self.SUBJECT_DOMAIN,\n                                               self.ROLES, self.PROJECT,\n                                               self.PROJECT_DOMAIN,\n                                               self.GROUPS)\n            assertion_xml = response.assertion.to_string()\n            # The expected values in the assertions bellow need to be \'str\' in\n            # Python 2 and \'bytes\' in Python 3\n            # make sure we have the proper tag and prefix for the assertion\n            # namespace\n            self.assertIn(b\'<saml:Assertion\', assertion_xml)\n            self.assertIn(\n                (\'xmlns:saml="\' + saml2.NAMESPACE + \'"\').encode(\'utf-8\'),\n                assertion_xml)\n            self.assertIn(\n                (\'xmlns:xmldsig="\' + xmldsig.NAMESPACE).encode(\'utf-8\'),\n                assertion_xml)\n\n    def test_saml_signing(self):\n        """Test that the SAML generator produces a SAML object.\n\n        Test the SAML generator directly by passing known arguments, the result\n        should be a SAML object that consistently includes attributes based on\n        the known arguments that were passed in.\n\n        """\n        if not _is_xmlsec1_installed():\n            self.skipTest(\'xmlsec1 is not installed\')\n\n        generator = keystone_idp.SAMLGenerator()\n        response = generator.samlize_token(self.ISSUER, self.RECIPIENT,\n                                           self.SUBJECT, self.SUBJECT_DOMAIN,\n                                           self.ROLES, self.PROJECT,\n                                           self.PROJECT_DOMAIN,\n                                           self.GROUPS)\n\n        signature = response.assertion.signature\n        self.assertIsNotNone(signature)\n        self.assertIsInstance(signature, xmldsig.Signature)\n\n        idp_public_key = sigver.read_cert_from_file(CONF.saml.certfile, \'pem\')\n        cert_text = signature.key_info.x509_data[0].x509_certificate.text\n        # NOTE(stevemar): Rather than one line of text, the certificate is\n        # printed with newlines for readability, we remove these so we can\n        # match it with the key that we used.\n        cert_text = cert_text.replace(os.linesep, \'\')\n        self.assertEqual(idp_public_key, cert_text)\n\n    def _create_generate_saml_request(self, token_id, sp_id):\n        return {\n            "auth": {\n                "identity": {\n                    "methods": [\n                        "token"\n                    ],\n                    "token": {\n                        "id": token_id\n                    }\n                },\n                "scope": {\n                    "service_provider": {\n                        "id": sp_id\n                    }\n                }\n            }\n        }\n\n    def _fetch_valid_token(self):\n        auth_data = self.build_authentication_request(\n            user_id=self.user[\'id\'],\n            password=self.user[\'password\'],\n            project_id=self.project[\'id\'])\n        resp = self.v3_create_token(auth_data)\n        token_id = resp.headers.get(\'X-Subject-Token\')\n        return token_id\n\n    def _fetch_domain_scoped_token(self):\n        auth_data = self.build_authentication_request(\n            user_id=self.user[\'id\'],\n            password=self.user[\'password\'],\n            user_domain_id=self.domain[\'id\'])\n        resp = self.v3_create_token(auth_data)\n        token_id = resp.headers.get(\'X-Subject-Token\')\n        return token_id\n\n    def test_not_project_scoped_token(self):\n        """Ensure SAML generation fails when passing domain-scoped tokens.\n\n        The server should return a 403 Forbidden Action.\n\n        """\n        self.config_fixture.config(group=\'saml\', idp_entity_id=self.ISSUER)\n        token_id = self._fetch_domain_scoped_token()\n        body = self._create_generate_saml_request(token_id,\n                                                  self.SERVICE_PROVDIER_ID)\n        with mock.patch.object(keystone_idp, \'_sign_assertion\',\n                               return_value=self.signed_assertion):\n            self.post(self.SAML_GENERATION_ROUTE, body=body,\n                      expected_status=http.client.FORBIDDEN)\n\n    def test_generate_saml_route(self):\n        """Test that the SAML generation endpoint produces XML.\n\n        The SAML endpoint /v3/auth/OS-FEDERATION/saml2 should take as input,\n        a scoped token ID, and a Service Provider ID.\n        The controller should fetch details about the user from the token,\n        and details about the service provider from its ID.\n        This should be enough information to invoke the SAML generator and\n        provide a valid SAML (XML) document back.\n\n        """\n        self.config_fixture.config(group=\'saml\', idp_entity_id=self.ISSUER)\n        token_id = self._fetch_valid_token()\n        body = self._create_generate_saml_request(token_id,\n                                                  self.SERVICE_PROVDIER_ID)\n\n        with mock.patch.object(keystone_idp, \'_sign_assertion\',\n                               return_value=self.signed_assertion):\n            http_response = self.post(self.SAML_GENERATION_ROUTE, body=body,\n                                      response_content_type=\'text/xml\',\n                                      expected_status=http.client.OK)\n\n        response = etree.fromstring(http_response.result)\n        issuer = response[0]\n        assertion = response[2]\n\n        self.assertEqual(self.RECIPIENT, response.get(\'Destination\'))\n        self.assertEqual(self.ISSUER, issuer.text)\n\n        # NOTE(stevemar): We should test this against expected values,\n        # but the self.xyz attribute names are uuids, and we mock out\n        # the result. Ideally we should update the mocked result with\n        # some known data, and create the roles/project/user before\n        # these tests run.\n        user_attribute = assertion[4][0]\n        self.assertIsInstance(user_attribute[0].text, str)\n\n        user_domain_attribute = assertion[4][1]\n        self.assertIsInstance(user_domain_attribute[0].text, str)\n\n        role_attribute = assertion[4][2]\n        self.assertIsInstance(role_attribute[0].text, str)\n\n        project_attribute = assertion[4][3]\n        self.assertIsInstance(project_attribute[0].text, str)\n\n        project_domain_attribute = assertion[4][4]\n        self.assertIsInstance(project_domain_attribute[0].text, str)\n\n        group_attribute = assertion[4][5]\n        self.assertIsInstance(group_attribute[0].text, str)\n\n    def test_invalid_scope_body(self):\n        """Test that missing the scope in request body raises an exception.\n\n        Raises exception.SchemaValidationError() - error 400 Bad Request\n\n        """\n        token_id = uuid.uuid4().hex\n        body = self._create_generate_saml_request(token_id,\n                                                  self.SERVICE_PROVDIER_ID)\n        del body[\'auth\'][\'scope\']\n\n        self.post(self.SAML_GENERATION_ROUTE, body=body,\n                  expected_status=http.client.BAD_REQUEST)\n\n    def test_invalid_token_body(self):\n        """Test that missing the token in request body raises an exception.\n\n        Raises exception.SchemaValidationError() - error 400 Bad Request\n\n        """\n        token_id = uuid.uuid4().hex\n        body = self._create_generate_saml_request(token_id,\n                                                  self.SERVICE_PROVDIER_ID)\n        del body[\'auth\'][\'identity\'][\'token\']\n\n        self.post(self.SAML_GENERATION_ROUTE, body=body,\n                  expected_status=http.client.BAD_REQUEST)\n\n    def test_sp_not_found(self):\n        """Test SAML generation with an invalid service provider ID.\n\n        Raises exception.ServiceProviderNotFound() - error Not Found 404\n\n        """\n        sp_id = uuid.uuid4().hex\n        token_id = self._fetch_valid_token()\n        body = self._create_generate_saml_request(token_id, sp_id)\n        self.post(self.SAML_GENERATION_ROUTE, body=body,\n                  expected_status=http.client.NOT_FOUND)\n\n    def test_sp_disabled(self):\n        """Try generating assertion for disabled Service Provider."""\n        # Disable Service Provider\n        sp_ref = {\'enabled\': False}\n        PROVIDERS.federation_api.update_sp(self.SERVICE_PROVDIER_ID, sp_ref)\n\n        token_id = self._fetch_valid_token()\n        body = self._create_generate_saml_request(token_id,\n                                                  self.SERVICE_PROVDIER_ID)\n        self.post(self.SAML_GENERATION_ROUTE, body=body,\n                  expected_status=http.client.FORBIDDEN)\n\n    def test_token_not_found(self):\n        """Test that an invalid token in the request body raises an exception.\n\n        Raises exception.TokenNotFound() - error Not Found 404\n\n        """\n        token_id = uuid.uuid4().hex\n        body = self._create_generate_saml_request(token_id,\n                                                  self.SERVICE_PROVDIER_ID)\n        self.post(self.SAML_GENERATION_ROUTE, body=body,\n                  expected_status=http.client.NOT_FOUND)\n\n    def test_generate_ecp_route(self):\n        """Test that the ECP generation endpoint produces XML.\n\n        The ECP endpoint /v3/auth/OS-FEDERATION/saml2/ecp should take the same\n        input as the SAML generation endpoint (scoped token ID + Service\n        Provider ID).\n        The controller should return a SAML assertion that is wrapped in a\n        SOAP envelope.\n        """\n        self.config_fixture.config(group=\'saml\', idp_entity_id=self.ISSUER)\n        token_id = self._fetch_valid_token()\n        body = self._create_generate_saml_request(token_id,\n                                                  self.SERVICE_PROVDIER_ID)\n\n        with mock.patch.object(keystone_idp, \'_sign_assertion\',\n                               return_value=self.signed_assertion):\n            http_response = self.post(self.ECP_GENERATION_ROUTE, body=body,\n                                      response_content_type=\'text/xml\',\n                                      expected_status=http.client.OK)\n\n        env_response = etree.fromstring(http_response.result)\n        header = env_response[0]\n\n        # Verify the relay state starts with \'ss:mem\'\n        prefix = CONF.saml.relay_state_prefix\n        self.assertThat(header[0].text, matchers.StartsWith(prefix))\n\n        # Verify that the content in the body matches the expected assertion\n        body = env_response[1]\n        response = body[0]\n        issuer = response[0]\n        assertion = response[2]\n\n        self.assertEqual(self.RECIPIENT, response.get(\'Destination\'))\n        self.assertEqual(self.ISSUER, issuer.text)\n\n        user_attribute = assertion[4][0]\n        self.assertIsInstance(user_attribute[0].text, str)\n\n        user_domain_attribute = assertion[4][1]\n        self.assertIsInstance(user_domain_attribute[0].text, str)\n\n        role_attribute = assertion[4][2]\n        self.assertIsInstance(role_attribute[0].text, str)\n\n        project_attribute = assertion[4][3]\n        self.assertIsInstance(project_attribute[0].text, str)\n\n        project_domain_attribute = assertion[4][4]\n        self.assertIsInstance(project_domain_attribute[0].text, str)\n\n        group_attribute = assertion[4][5]\n        self.assertIsInstance(group_attribute[0].text, str)\n\n    @mock.patch(\'saml2.create_class_from_xml_string\')\n    @mock.patch(\'oslo_utils.fileutils.write_to_tempfile\')\n    @mock.patch.object(subprocess, \'check_output\')\n    def test_sign_assertion(self, check_output_mock,\n                            write_to_tempfile_mock, create_class_mock):\n        write_to_tempfile_mock.return_value = \'tmp_path\'\n        check_output_mock.return_value = \'fakeoutput\'\n\n        keystone_idp._sign_assertion(self.signed_assertion)\n\n        create_class_mock.assert_called_with(saml.Assertion, \'fakeoutput\')\n\n    @mock.patch(\'oslo_utils.fileutils.write_to_tempfile\')\n    def test_sign_assertion_exc(self, write_to_tempfile_mock):\n        # If the command fails the command output is logged.\n        sample_returncode = 1\n        sample_output = self.getUniqueString()\n        write_to_tempfile_mock.return_value = \'tmp_path\'\n\n        def side_effect(*args, **kwargs):\n            if args[0] == [\'/usr/bin/which\', CONF.saml.xmlsec1_binary]:\n                return \'/usr/bin/xmlsec1\\n\'\n            else:\n                raise subprocess.CalledProcessError(\n                    returncode=sample_returncode, cmd=CONF.saml.xmlsec1_binary,\n                    output=sample_output\n                )\n\n        with mock.patch.object(subprocess, \'check_output\',\n                               side_effect=side_effect):\n            logger_fixture = self.useFixture(fixtures.LoggerFixture())\n            self.assertRaises(\n                exception.SAMLSigningError,\n                keystone_idp._sign_assertion,\n                self.signed_assertion\n            )\n\n            # The function __str__ in subprocess.CalledProcessError is\n            # different between py3.6 and lower python version.\n            expected_log = (\n                r"Error when signing assertion, reason: Command \'%s\' returned "\n                r"non-zero exit status %s\\.? %s\\n" %\n                (CONF.saml.xmlsec1_binary, sample_returncode, sample_output))\n            self.assertRegex(logger_fixture.output,\n                             re.compile(r\'%s\' % expected_log))\n\n    @mock.patch(\'oslo_utils.fileutils.write_to_tempfile\')\n    @mock.patch.object(subprocess, \'check_output\')\n    def test_sign_assertion_fileutils_exc(self, check_output_mock,\n                                          write_to_tempfile_mock):\n        exception_msg = \'fake\'\n        write_to_tempfile_mock.side_effect = Exception(exception_msg)\n        check_output_mock.return_value = \'/usr/bin/xmlsec1\'\n\n        logger_fixture = self.useFixture(fixtures.LoggerFixture())\n        self.assertRaises(exception.SAMLSigningError,\n                          keystone_idp._sign_assertion,\n                          self.signed_assertion)\n        expected_log = (\n            \'Error when signing assertion, reason: %s\\n\' % exception_msg)\n        self.assertEqual(expected_log, logger_fixture.output)\n\n    def test_sign_assertion_logs_message_if_xmlsec1_is_not_installed(self):\n        with mock.patch.object(subprocess, \'check_output\') as co_mock:\n            co_mock.side_effect = subprocess.CalledProcessError(\n                returncode=1, cmd=CONF.saml.xmlsec1_binary,\n            )\n            logger_fixture = self.useFixture(fixtures.LoggerFixture())\n            self.assertRaises(\n                exception.SAMLSigningError,\n                keystone_idp._sign_assertion,\n                self.signed_assertion\n            )\n\n            expected_log = (\'Unable to locate xmlsec1 binary on the system. \'\n                            \'Check to make sure it is installed.\\n\')\n            self.assertEqual(expected_log, logger_fixture.output)\n\n\nclass IdPMetadataGenerationTests(test_v3.RestfulTestCase):\n    """A class for testing Identity Provider Metadata generation."""\n\n    METADATA_URL = \'/OS-FEDERATION/saml2/metadata\'\n\n    def setUp(self):\n        super(IdPMetadataGenerationTests, self).setUp()\n        self.generator = keystone_idp.MetadataGenerator()\n\n    def config_overrides(self):\n        super(IdPMetadataGenerationTests, self).config_overrides()\n        self.config_fixture.config(\n            group=\'saml\',\n            idp_entity_id=federation_fixtures.IDP_ENTITY_ID,\n            idp_sso_endpoint=federation_fixtures.IDP_SSO_ENDPOINT,\n            idp_organization_name=federation_fixtures.IDP_ORGANIZATION_NAME,\n            idp_organization_display_name=(\n                federation_fixtures.IDP_ORGANIZATION_DISPLAY_NAME),\n            idp_organization_url=federation_fixtures.IDP_ORGANIZATION_URL,\n            idp_contact_company=federation_fixtures.IDP_CONTACT_COMPANY,\n            idp_contact_name=federation_fixtures.IDP_CONTACT_GIVEN_NAME,\n            idp_contact_surname=federation_fixtures.IDP_CONTACT_SURNAME,\n            idp_contact_email=federation_fixtures.IDP_CONTACT_EMAIL,\n            idp_contact_telephone=(\n                federation_fixtures.IDP_CONTACT_TELEPHONE_NUMBER),\n            idp_contact_type=federation_fixtures.IDP_CONTACT_TYPE)\n\n    def test_check_entity_id(self):\n        metadata = self.generator.generate_metadata()\n        self.assertEqual(federation_fixtures.IDP_ENTITY_ID, metadata.entity_id)\n\n    def test_metadata_validity(self):\n        """Call md.EntityDescriptor method that does internal verification."""\n        self.generator.generate_metadata().verify()\n\n    def test_serialize_metadata_object(self):\n        """Check whether serialization doesn\'t raise any exceptions."""\n        self.generator.generate_metadata().to_string()\n        # TODO(marek-denis): Check values here\n\n    def test_check_idp_sso(self):\n        metadata = self.generator.generate_metadata()\n        idpsso_descriptor = metadata.idpsso_descriptor\n        self.assertIsNotNone(metadata.idpsso_descriptor)\n        self.assertEqual(federation_fixtures.IDP_SSO_ENDPOINT,\n                         idpsso_descriptor.single_sign_on_service.location)\n\n        self.assertIsNotNone(idpsso_descriptor.organization)\n        organization = idpsso_descriptor.organization\n        self.assertEqual(federation_fixtures.IDP_ORGANIZATION_DISPLAY_NAME,\n                         organization.organization_display_name.text)\n        self.assertEqual(federation_fixtures.IDP_ORGANIZATION_NAME,\n                         organization.organization_name.text)\n        self.assertEqual(federation_fixtures.IDP_ORGANIZATION_URL,\n                         organization.organization_url.text)\n\n        self.assertIsNotNone(idpsso_descriptor.contact_person)\n        contact_person = idpsso_descriptor.contact_person\n\n        self.assertEqual(federation_fixtures.IDP_CONTACT_GIVEN_NAME,\n                         contact_person.given_name.text)\n        self.assertEqual(federation_fixtures.IDP_CONTACT_SURNAME,\n                         contact_person.sur_name.text)\n        self.assertEqual(federation_fixtures.IDP_CONTACT_EMAIL,\n                         contact_person.email_address.text)\n        self.assertEqual(federation_fixtures.IDP_CONTACT_TELEPHONE_NUMBER,\n                         contact_person.telephone_number.text)\n        self.assertEqual(federation_fixtures.IDP_CONTACT_TYPE,\n                         contact_person.contact_type)\n\n    def test_metadata_no_organization(self):\n        self.config_fixture.config(\n            group=\'saml\',\n            idp_organization_display_name=None,\n            idp_organization_url=None,\n            idp_organization_name=None)\n        metadata = self.generator.generate_metadata()\n        idpsso_descriptor = metadata.idpsso_descriptor\n        self.assertIsNotNone(metadata.idpsso_descriptor)\n        self.assertIsNone(idpsso_descriptor.organization)\n        self.assertIsNotNone(idpsso_descriptor.contact_person)\n\n    def test_metadata_no_contact_person(self):\n        self.config_fixture.config(\n            group=\'saml\',\n            idp_contact_name=None,\n            idp_contact_surname=None,\n            idp_contact_email=None,\n            idp_contact_telephone=None)\n        metadata = self.generator.generate_metadata()\n        idpsso_descriptor = metadata.idpsso_descriptor\n        self.assertIsNotNone(metadata.idpsso_descriptor)\n        self.assertIsNotNone(idpsso_descriptor.organization)\n        self.assertEqual([], idpsso_descriptor.contact_person)\n\n    def test_metadata_invalid_idp_sso_endpoint(self):\n        self.config_fixture.config(\n            group=\'saml\',\n            idp_sso_endpoint=None)\n        self.assertRaises(exception.ValidationError,\n                          self.generator.generate_metadata)\n\n    def test_metadata_invalid_idp_entity_id(self):\n        self.config_fixture.config(\n            group=\'saml\',\n            idp_entity_id=None)\n        self.assertRaises(exception.ValidationError,\n                          self.generator.generate_metadata)\n\n    def test_get_metadata_with_no_metadata_file_configured(self):\n        self.get(self.METADATA_URL,\n                 expected_status=http.client.INTERNAL_SERVER_ERROR)\n\n    def test_get_head_metadata(self):\n        self.config_fixture.config(\n            group=\'saml\', idp_metadata_path=XMLDIR + \'/idp_saml2_metadata.xml\')\n        self.head(self.METADATA_URL, expected_status=http.client.OK)\n        r = self.get(self.METADATA_URL, response_content_type=\'text/xml\')\n        self.assertEqual(\'text/xml\', r.headers.get(\'Content-Type\'))\n\n        reference_file = _load_xml(\'idp_saml2_metadata.xml\')\n\n        # `reference_file` needs to be converted to bytes to be able to be\n        # compared to `r.result` in the case of Python 3.\n        reference_file = str.encode(reference_file)\n        self.assertEqual(reference_file, r.result)\n\n\nclass ServiceProviderTests(test_v3.RestfulTestCase):\n    """A test class for Service Providers."""\n\n    MEMBER_NAME = \'service_provider\'\n    COLLECTION_NAME = \'service_providers\'\n    SERVICE_PROVIDER_ID = \'ACME\'\n    SP_KEYS = [\'auth_url\', \'id\', \'enabled\', \'description\',\n               \'relay_state_prefix\', \'sp_url\']\n\n    def setUp(self):\n        super(ServiceProviderTests, self).setUp()\n        # Add a Service Provider\n        url = self.base_url(suffix=self.SERVICE_PROVIDER_ID)\n        self.SP_REF = core.new_service_provider_ref()\n        self.SERVICE_PROVIDER = self.put(\n            url, body={\'service_provider\': self.SP_REF},\n            expected_status=http.client.CREATED).result\n\n    def base_url(self, suffix=None):\n        if suffix is not None:\n            return \'/OS-FEDERATION/service_providers/\' + str(suffix)\n        return \'/OS-FEDERATION/service_providers\'\n\n    def _create_default_sp(self, body=None):\n        """Create default Service Provider."""\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        if body is None:\n            body = core.new_service_provider_ref()\n        resp = self.put(url, body={\'service_provider\': body},\n                        expected_status=http.client.CREATED)\n        return resp\n\n    def test_get_head_service_provider(self):\n        url = self.base_url(suffix=self.SERVICE_PROVIDER_ID)\n        resp = self.get(url)\n        self.assertValidEntity(resp.result[\'service_provider\'],\n                               keys_to_check=self.SP_KEYS)\n        resp = self.head(url, expected_status=http.client.OK)\n\n    def test_get_service_provider_fail(self):\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        self.get(url, expected_status=http.client.NOT_FOUND)\n\n    def test_create_service_provider(self):\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        sp = core.new_service_provider_ref()\n        resp = self.put(url, body={\'service_provider\': sp},\n                        expected_status=http.client.CREATED)\n        self.assertValidEntity(resp.result[\'service_provider\'],\n                               keys_to_check=self.SP_KEYS)\n\n    @unit.skip_if_cache_disabled(\'federation\')\n    def test_create_service_provider_invalidates_cache(self):\n        # List all service providers and make sure we only have one in the\n        # list. This service provider is from testing setup.\n        resp = self.get(self.base_url(), expected_status=http.client.OK)\n        self.assertThat(\n            resp.json_body[\'service_providers\'],\n            matchers.HasLength(1)\n        )\n\n        # Create a new service provider.\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        sp = core.new_service_provider_ref()\n        self.put(url, body={\'service_provider\': sp},\n                 expected_status=http.client.CREATED)\n\n        # List all service providers again and make sure we have two in the\n        # returned list.\n        resp = self.get(self.base_url(), expected_status=http.client.OK)\n        self.assertThat(\n            resp.json_body[\'service_providers\'],\n            matchers.HasLength(2)\n        )\n\n    @unit.skip_if_cache_disabled(\'federation\')\n    def test_delete_service_provider_invalidates_cache(self):\n        # List all service providers and make sure we only have one in the\n        # list. This service provider is from testing setup.\n        resp = self.get(self.base_url(), expected_status=http.client.OK)\n        self.assertThat(\n            resp.json_body[\'service_providers\'],\n            matchers.HasLength(1)\n        )\n\n        # Create a new service provider.\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        sp = core.new_service_provider_ref()\n        self.put(url, body={\'service_provider\': sp},\n                 expected_status=http.client.CREATED)\n\n        # List all service providers again and make sure we have two in the\n        # returned list.\n        resp = self.get(self.base_url(), expected_status=http.client.OK)\n        self.assertThat(\n            resp.json_body[\'service_providers\'],\n            matchers.HasLength(2)\n        )\n\n        # Delete the service provider we created, which should invalidate the\n        # service provider cache. Get the list of service providers again and\n        # if the cache invalidated properly then we should only have one\n        # service provider in the list.\n        self.delete(url, expected_status=http.client.NO_CONTENT)\n        resp = self.get(self.base_url(), expected_status=http.client.OK)\n        self.assertThat(\n            resp.json_body[\'service_providers\'],\n            matchers.HasLength(1)\n        )\n\n    @unit.skip_if_cache_disabled(\'federation\')\n    def test_update_service_provider_invalidates_cache(self):\n        # List all service providers and make sure we only have one in the\n        # list. This service provider is from testing setup.\n        resp = self.get(self.base_url(), expected_status=http.client.OK)\n        self.assertThat(\n            resp.json_body[\'service_providers\'],\n            matchers.HasLength(1)\n        )\n\n        # Create a new service provider.\n        service_provider_id = uuid.uuid4().hex\n        url = self.base_url(suffix=service_provider_id)\n        sp = core.new_service_provider_ref()\n        self.put(url, body={\'service_provider\': sp},\n                 expected_status=http.client.CREATED)\n\n        # List all service providers again and make sure we have two in the\n        # returned list.\n        resp = self.get(self.base_url(), expected_status=http.client.OK)\n        self.assertThat(\n            resp.json_body[\'service_providers\'],\n            matchers.HasLength(2)\n        )\n\n        # Update the service provider we created, which should invalidate the\n        # service provider cache. Get the list of service providers again and\n        # if the cache invalidated properly then we see the value we updated.\n        updated_description = uuid.uuid4().hex\n        body = {\'service_provider\': {\'description\': updated_description}}\n        self.patch(url, body=body, expected_status=http.client.OK)\n        resp = self.get(self.base_url(), expected_status=http.client.OK)\n        self.assertThat(\n            resp.json_body[\'service_providers\'],\n            matchers.HasLength(2)\n        )\n        for sp in resp.json_body[\'service_providers\']:\n            if sp[\'id\'] == service_provider_id:\n                self.assertEqual(sp[\'description\'], updated_description)\n\n    def test_create_sp_relay_state_default(self):\n        """Create an SP without relay state, should default to `ss:mem`."""\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        sp = core.new_service_provider_ref()\n        del sp[\'relay_state_prefix\']\n        resp = self.put(url, body={\'service_provider\': sp},\n                        expected_status=http.client.CREATED)\n        sp_result = resp.result[\'service_provider\']\n        self.assertEqual(CONF.saml.relay_state_prefix,\n                         sp_result[\'relay_state_prefix\'])\n\n    def test_create_sp_relay_state_non_default(self):\n        """Create an SP with custom relay state."""\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        sp = core.new_service_provider_ref()\n        non_default_prefix = uuid.uuid4().hex\n        sp[\'relay_state_prefix\'] = non_default_prefix\n        resp = self.put(url, body={\'service_provider\': sp},\n                        expected_status=http.client.CREATED)\n        sp_result = resp.result[\'service_provider\']\n        self.assertEqual(non_default_prefix,\n                         sp_result[\'relay_state_prefix\'])\n\n    def test_create_service_provider_fail(self):\n        """Try adding SP object with unallowed attribute."""\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        sp = core.new_service_provider_ref()\n        sp[uuid.uuid4().hex] = uuid.uuid4().hex\n        self.put(url, body={\'service_provider\': sp},\n                 expected_status=http.client.BAD_REQUEST)\n\n    def test_list_head_service_providers(self):\n        """Test listing of service provider objects.\n\n        Add two new service providers. List all available service providers.\n        Expect to get list of three service providers (one created by setUp())\n        Test if attributes match.\n\n        """\n        ref_service_providers = {\n            uuid.uuid4().hex: core.new_service_provider_ref(),\n            uuid.uuid4().hex: core.new_service_provider_ref(),\n        }\n        for id, sp in ref_service_providers.items():\n            url = self.base_url(suffix=id)\n            self.put(url, body={\'service_provider\': sp},\n                     expected_status=http.client.CREATED)\n\n        # Insert ids into service provider object, we will compare it with\n        # responses from server and those include \'id\' attribute.\n\n        ref_service_providers[self.SERVICE_PROVIDER_ID] = self.SP_REF\n        for id, sp in ref_service_providers.items():\n            sp[\'id\'] = id\n\n        url = self.base_url()\n        resp = self.get(url)\n        service_providers = resp.result\n        for service_provider in service_providers[\'service_providers\']:\n            id = service_provider[\'id\']\n            self.assertValidEntity(\n                service_provider, ref=ref_service_providers[id],\n                keys_to_check=self.SP_KEYS)\n\n        self.head(url, expected_status=http.client.OK)\n\n    def test_update_service_provider(self):\n        """Update existing service provider.\n\n        Update default existing service provider and make sure it has been\n        properly changed.\n\n        """\n        new_sp_ref = core.new_service_provider_ref()\n        url = self.base_url(suffix=self.SERVICE_PROVIDER_ID)\n        resp = self.patch(url, body={\'service_provider\': new_sp_ref})\n        patch_result = resp.result\n        new_sp_ref[\'id\'] = self.SERVICE_PROVIDER_ID\n        self.assertValidEntity(patch_result[\'service_provider\'],\n                               ref=new_sp_ref,\n                               keys_to_check=self.SP_KEYS)\n\n        resp = self.get(url)\n        get_result = resp.result\n\n        self.assertDictEqual(patch_result[\'service_provider\'],\n                             get_result[\'service_provider\'])\n\n    def test_update_service_provider_immutable_parameters(self):\n        """Update immutable attributes in service provider.\n\n        In this particular case the test will try to change ``id`` attribute.\n        The server should return an HTTP 403 Forbidden error code.\n\n        """\n        new_sp_ref = {\'id\': uuid.uuid4().hex}\n        url = self.base_url(suffix=self.SERVICE_PROVIDER_ID)\n        self.patch(url, body={\'service_provider\': new_sp_ref},\n                   expected_status=http.client.BAD_REQUEST)\n\n    def test_update_service_provider_unknown_parameter(self):\n        new_sp_ref = core.new_service_provider_ref()\n        new_sp_ref[uuid.uuid4().hex] = uuid.uuid4().hex\n        url = self.base_url(suffix=self.SERVICE_PROVIDER_ID)\n        self.patch(url, body={\'service_provider\': new_sp_ref},\n                   expected_status=http.client.BAD_REQUEST)\n\n    def test_update_service_provider_returns_not_found(self):\n        new_sp_ref = core.new_service_provider_ref()\n        new_sp_ref[\'description\'] = uuid.uuid4().hex\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        self.patch(url, body={\'service_provider\': new_sp_ref},\n                   expected_status=http.client.NOT_FOUND)\n\n    def test_update_sp_relay_state(self):\n        """Update an SP with custom relay state."""\n        new_sp_ref = core.new_service_provider_ref()\n        non_default_prefix = uuid.uuid4().hex\n        new_sp_ref[\'relay_state_prefix\'] = non_default_prefix\n        url = self.base_url(suffix=self.SERVICE_PROVIDER_ID)\n        resp = self.patch(url, body={\'service_provider\': new_sp_ref})\n        sp_result = resp.result[\'service_provider\']\n        self.assertEqual(non_default_prefix,\n                         sp_result[\'relay_state_prefix\'])\n\n    def test_delete_service_provider(self):\n        url = self.base_url(suffix=self.SERVICE_PROVIDER_ID)\n        self.delete(url)\n\n    def test_delete_service_provider_returns_not_found(self):\n        url = self.base_url(suffix=uuid.uuid4().hex)\n        self.delete(url, expected_status=http.client.NOT_FOUND)\n\n    def test_filter_list_sp_by_id(self):\n        def get_id(resp):\n            sp = resp.result.get(\'service_provider\')\n            return sp.get(\'id\')\n\n        sp1_id = get_id(self._create_default_sp())\n        sp2_id = get_id(self._create_default_sp())\n\n        # list the SP, should get SPs.\n        url = self.base_url()\n        resp = self.get(url)\n        sps = resp.result.get(\'service_providers\')\n        entities_ids = [e[\'id\'] for e in sps]\n        self.assertIn(sp1_id, entities_ids)\n        self.assertIn(sp2_id, entities_ids)\n\n        # filter the SP by \'id\'. Only SP1 should appear.\n        url = self.base_url() + \'?id=\' + sp1_id\n        resp = self.get(url)\n        sps = resp.result.get(\'service_providers\')\n        entities_ids = [e[\'id\'] for e in sps]\n        self.assertIn(sp1_id, entities_ids)\n        self.assertNotIn(sp2_id, entities_ids)\n\n    def test_filter_list_sp_by_enabled(self):\n        def get_id(resp):\n            sp = resp.result.get(\'service_provider\')\n            return sp.get(\'id\')\n\n        sp1_id = get_id(self._create_default_sp())\n        sp2_ref = core.new_service_provider_ref()\n        sp2_ref[\'enabled\'] = False\n        sp2_id = get_id(self._create_default_sp(body=sp2_ref))\n\n        # list the SP, should get two SPs.\n        url = self.base_url()\n        resp = self.get(url)\n        sps = resp.result.get(\'service_providers\')\n        entities_ids = [e[\'id\'] for e in sps]\n        self.assertIn(sp1_id, entities_ids)\n        self.assertIn(sp2_id, entities_ids)\n\n        # filter the SP by \'enabled\'. Only SP1 should appear.\n        url = self.base_url() + \'?enabled=True\'\n        resp = self.get(url)\n        sps = resp.result.get(\'service_providers\')\n        entities_ids = [e[\'id\'] for e in sps]\n        self.assertIn(sp1_id, entities_ids)\n        self.assertNotIn(sp2_id, entities_ids)\n\n\nclass WebSSOTests(FederatedTokenTests):\n    """A class for testing Web SSO."""\n\n    SSO_URL = \'/auth/OS-FEDERATION/websso/\'\n    SSO_TEMPLATE_NAME = \'sso_callback_template.html\'\n    SSO_TEMPLATE_PATH = os.path.join(core.dirs.etc(), SSO_TEMPLATE_NAME)\n    TRUSTED_DASHBOARD = \'http://horizon.com\'\n    ORIGIN = urllib.parse.quote_plus(TRUSTED_DASHBOARD)\n    PROTOCOL_REMOTE_ID_ATTR = uuid.uuid4().hex\n\n    def config_overrides(self):\n        super(WebSSOTests, self).config_overrides()\n        self.config_fixture.config(\n            group=\'federation\',\n            trusted_dashboard=[self.TRUSTED_DASHBOARD],\n            sso_callback_template=self.SSO_TEMPLATE_PATH,\n            remote_id_attribute=self.REMOTE_ID_ATTR)\n\n    def test_render_callback_template(self):\n        token_id = uuid.uuid4().hex\n        with self.make_request():\n            resp = (\n                auth_api._AuthFederationWebSSOBase._render_template_response(\n                    self.TRUSTED_DASHBOARD, token_id))\n        # The expected value in the assertions bellow need to be \'str\' in\n        # Python 2 and \'bytes\' in Python 3\n        self.assertIn(token_id.encode(\'utf-8\'), resp.data)\n        self.assertIn(self.TRUSTED_DASHBOARD.encode(\'utf-8\'), resp.data)\n\n    def test_federated_sso_auth(self):\n        environment = {self.REMOTE_ID_ATTR: self.REMOTE_IDS[0],\n                       \'QUERY_STRING\': \'origin=%s\' % self.ORIGIN}\n        environment.update(mapping_fixtures.EMPLOYEE_ASSERTION)\n        with self.make_request(environ=environment):\n            resp = auth_api.AuthFederationWebSSOResource._perform_auth(\n                self.PROTOCOL)\n        # `resp.data` will be `str` in Python 2 and `bytes` in Python 3\n        # which is why expected value: `self.TRUSTED_DASHBOARD`\n        # needs to be encoded\n        self.assertIn(self.TRUSTED_DASHBOARD.encode(\'utf-8\'), resp.data)\n\n    def test_get_sso_origin_host_case_insensitive(self):\n        # test lowercase hostname in trusted_dashboard\n        environ = {\'QUERY_STRING\': \'origin=http://horizon.com\'}\n        with self.make_request(environ=environ):\n            host = auth_api._get_sso_origin_host()\n            self.assertEqual("http://horizon.com", host)\n            # test uppercase hostname in trusted_dashboard\n            self.config_fixture.config(\n                group=\'federation\',\n                trusted_dashboard=[\'http://Horizon.com\'])\n            host = auth_api._get_sso_origin_host()\n            self.assertEqual("http://horizon.com", host)\n\n    def test_federated_sso_auth_with_protocol_specific_remote_id(self):\n        self.config_fixture.config(\n            group=self.PROTOCOL,\n            remote_id_attribute=self.PROTOCOL_REMOTE_ID_ATTR)\n\n        environment = {self.PROTOCOL_REMOTE_ID_ATTR: self.REMOTE_IDS[0],\n                       \'QUERY_STRING\': \'origin=%s\' % self.ORIGIN}\n        environment.update(mapping_fixtures.EMPLOYEE_ASSERTION)\n        with self.make_request(environ=environment):\n            resp = auth_api.AuthFederationWebSSOResource._perform_auth(\n                self.PROTOCOL)\n        # `resp.data` will be `str` in Python 2 and `bytes` in Python 3\n        # which is why expected value: `self.TRUSTED_DASHBOARD`\n        # needs to be encoded\n        self.assertIn(self.TRUSTED_DASHBOARD.encode(\'utf-8\'), resp.data)\n\n    def test_federated_sso_auth_bad_remote_id(self):\n        environment = {self.REMOTE_ID_ATTR: self.IDP,\n                       \'QUERY_STRING\': \'origin=%s\' % self.ORIGIN}\n        environment.update(mapping_fixtures.EMPLOYEE_ASSERTION)\n        with self.make_request(environ=environment):\n            self.assertRaises(\n                exception.IdentityProviderNotFound,\n                auth_api.AuthFederationWebSSOResource._perform_auth,\n                self.PROTOCOL)\n\n    def test_federated_sso_missing_query(self):\n        environment = {self.REMOTE_ID_ATTR: self.REMOTE_IDS[0]}\n        environment.update(mapping_fixtures.EMPLOYEE_ASSERTION)\n        with self.make_request(environ=environment):\n            self.assertRaises(\n                exception.ValidationError,\n                auth_api.AuthFederationWebSSOResource._perform_auth,\n                self.PROTOCOL)\n\n    def test_federated_sso_missing_query_bad_remote_id(self):\n        environment = {self.REMOTE_ID_ATTR: self.IDP}\n        environment.update(mapping_fixtures.EMPLOYEE_ASSERTION)\n        with self.make_request(environ=environment):\n            self.assertRaises(\n                exception.ValidationError,\n                auth_api.AuthFederationWebSSOResource._perform_auth,\n                self.PROTOCOL)\n\n    def test_federated_sso_auth_protocol_not_found(self):\n        environment = {self.REMOTE_ID_ATTR: self.REMOTE_IDS[0],\n                       \'QUERY_STRING\': \'origin=%s\' % self.ORIGIN}\n        environment.update(mapping_fixtures.EMPLOYEE_ASSERTION)\n        with self.make_request(environ=environment):\n            self.assertRaises(\n                exception.Unauthorized,\n                auth_api.AuthFederationWebSSOResource._perform_auth,\n                \'no_this_protocol\')\n\n    def test_federated_sso_untrusted_dashboard(self):\n        environment = {self.REMOTE_ID_ATTR: self.REMOTE_IDS[0],\n                       \'QUERY_STRING\': \'origin=%s\' % uuid.uuid4().hex}\n        environment.update(mapping_fixtures.EMPLOYEE_ASSERTION)\n        with self.make_request(environ=environment):\n            self.assertRaises(\n                exception.Unauthorized,\n                auth_api.AuthFederationWebSSOResource._perform_auth,\n                self.PROTOCOL)\n\n    def test_federated_sso_untrusted_dashboard_bad_remote_id(self):\n        environment = {self.REMOTE_ID_ATTR: self.IDP,\n                       \'QUERY_STRING\': \'origin=%s\' % uuid.uuid4().hex}\n        environment.update(mapping_fixtures.EMPLOYEE_ASSERTION)\n        with self.make_request(environ=environment):\n            self.assertRaises(\n                exception.Unauthorized,\n                auth_api.AuthFederationWebSSOResource._perform_auth,\n                self.PROTOCOL)\n\n    def test_federated_sso_missing_remote_id(self):\n        environment = copy.deepcopy(mapping_fixtures.EMPLOYEE_ASSERTION)\n        with self.make_request(environ=environment,\n                               query_string=\'origin=%s\' % self.ORIGIN):\n            self.assertRaises(\n                exception.Unauthorized,\n                auth_api.AuthFederationWebSSOResource._perform_auth,\n                self.PROTOCOL)\n\n    def test_identity_provider_specific_federated_authentication(self):\n        environment = {self.REMOTE_ID_ATTR: self.REMOTE_IDS[0]}\n        environment.update(mapping_fixtures.EMPLOYEE_ASSERTION)\n        with self.make_request(environ=environment,\n                               query_string=\'origin=%s\' % self.ORIGIN):\n            resp = auth_api.AuthFederationWebSSOIDPsResource._perform_auth(\n                self.idp[\'id\'], self.PROTOCOL)\n        # `resp.data` will be `str` in Python 2 and `bytes` in Python 3\n        # which is why the expected value: `self.TRUSTED_DASHBOARD`\n        # needs to be encoded\n        self.assertIn(self.TRUSTED_DASHBOARD.encode(\'utf-8\'), resp.data)\n\n    def test_issue_unscoped_token_with_remote_from_protocol(self):\n        self.config_fixture.config(\n            group=\'federation\', remote_id_attribute=None\n        )\n        self.config_fixture.config(\n            group=self.PROTOCOL, remote_id_attribute=None\n        )\n        protocol = PROVIDERS.federation_api.get_protocol(\n            self.IDP_WITH_REMOTE, self.PROTOCOL\n        )\n        protocol[\'remote_id_attribute\'] = self.PROTOCOL_REMOTE_ID_ATTR\n        PROVIDERS.federation_api.update_protocol(\n            self.IDP_WITH_REMOTE, protocol[\'id\'], protocol\n        )\n        environment = {self.PROTOCOL_REMOTE_ID_ATTR: self.REMOTE_IDS[0],\n                       \'QUERY_STRING\': \'origin=%s\' % self.ORIGIN}\n        environment.update(mapping_fixtures.EMPLOYEE_ASSERTION)\n        with self.make_request(environ=environment):\n            resp = auth_api.AuthFederationWebSSOResource._perform_auth(\n                self.PROTOCOL)\n        self.assertIn(self.TRUSTED_DASHBOARD.encode(\'utf-8\'), resp.data)\n\n\nclass K2KServiceCatalogTests(test_v3.RestfulTestCase):\n    SP1 = \'SP1\'\n    SP2 = \'SP2\'\n    SP3 = \'SP3\'\n\n    def setUp(self):\n        super(K2KServiceCatalogTests, self).setUp()\n\n        sp = core.new_service_provider_ref()\n        PROVIDERS.federation_api.create_sp(self.SP1, sp)\n        self.sp_alpha = {self.SP1: sp}\n\n        sp = core.new_service_provider_ref()\n        PROVIDERS.federation_api.create_sp(self.SP2, sp)\n        self.sp_beta = {self.SP2: sp}\n\n        sp = core.new_service_provider_ref()\n        PROVIDERS.federation_api.create_sp(self.SP3, sp)\n        self.sp_gamma = {self.SP3: sp}\n\n    def sp_response(self, id, ref):\n        ref.pop(\'enabled\')\n        ref.pop(\'description\')\n        ref.pop(\'relay_state_prefix\')\n        ref[\'id\'] = id\n        return ref\n\n    def _validate_service_providers(self, token, ref):\n        token_data = token[\'token\']\n        self.assertIn(\'service_providers\', token_data)\n        self.assertIsNotNone(token_data[\'service_providers\'])\n        service_providers = token_data.get(\'service_providers\')\n\n        self.assertEqual(len(ref), len(service_providers))\n        for entity in service_providers:\n            id = entity.get(\'id\')\n            ref_entity = self.sp_response(id, ref.get(id))\n            self.assertDictEqual(entity, ref_entity)\n\n    def test_service_providers_in_token(self):\n        """Check if service providers are listed in service catalog."""\n        model = token_model.TokenModel()\n        model.user_id = self.user_id\n        model.methods = [\'password\']\n        token = render_token.render_token_response_from_model(model)\n        ref = {}\n        for r in (self.sp_alpha, self.sp_beta, self.sp_gamma):\n            ref.update(r)\n        self._validate_service_providers(token, ref)\n\n    def test_service_provides_in_token_disabled_sp(self):\n        """Test behaviour with disabled service providers.\n\n        Disabled service providers should not be listed in the service\n        catalog.\n\n        """\n        # disable service provider ALPHA\n        sp_ref = {\'enabled\': False}\n        PROVIDERS.federation_api.update_sp(self.SP1, sp_ref)\n\n        model = token_model.TokenModel()\n        model.user_id = self.user_id\n        model.methods = [\'password\']\n        token = render_token.render_token_response_from_model(model)\n        ref = {}\n        for r in (self.sp_beta, self.sp_gamma):\n            ref.update(r)\n        self._validate_service_providers(token, ref)\n\n    def test_no_service_providers_in_token(self):\n        """Test service catalog with disabled service providers.\n\n        There should be no entry ``service_providers`` in the catalog.\n        Test passes providing no attribute was raised.\n\n        """\n        sp_ref = {\'enabled\': False}\n        for sp in (self.SP1, self.SP2, self.SP3):\n            PROVIDERS.federation_api.update_sp(sp, sp_ref)\n\n        model = token_model.TokenModel()\n        model.user_id = self.user_id\n        model.methods = [\'password\']\n        token = render_token.render_token_response_from_model(model)\n        self.assertNotIn(\'service_providers\', token[\'token\'],\n                         message=(\'Expected Service Catalog not to have \'\n                                  \'service_providers\'))\n', 'message': 'Merge "Refactor: change 400 status codes in test names"', 'author_name': 'Gerrit Code Review', 'lang': 'py'}