# 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 uuid import six from keystoneauth1 import access from keystoneauth1.auth.identity import v3 from keystoneauth1 import exceptions from keystoneauth1 import fixture from keystoneauth1 import session from keystoneauth1.tests.unit import k2k_fixtures from keystoneauth1.tests.unit import utils class TesterFederationPlugin(v3.FederationBaseAuth): def get_unscoped_auth_ref(self, sess, **kwargs): # This would go and talk to an idp or something resp = sess.post(self.federated_token_url, authenticated=False) return access.create(resp=resp) class V3FederatedPlugin(utils.TestCase): AUTH_URL = 'http://keystone/v3' def setUp(self): super(V3FederatedPlugin, self).setUp() self.unscoped_token = fixture.V3Token() self.unscoped_token_id = uuid.uuid4().hex self.scoped_token = copy.deepcopy(self.unscoped_token) self.scoped_token.set_project_scope() self.scoped_token.methods.append('token') self.scoped_token_id = uuid.uuid4().hex s = self.scoped_token.add_service('compute', name='nova') s.add_standard_endpoints(public='http://nova/public', admin='http://nova/admin', internal='http://nova/internal') self.idp = uuid.uuid4().hex self.protocol = uuid.uuid4().hex self.token_url = ('%s/OS-FEDERATION/identity_providers/%s/protocols/%s' '/auth' % (self.AUTH_URL, self.idp, self.protocol)) headers = {'X-Subject-Token': self.unscoped_token_id} self.unscoped_mock = self.requests_mock.post(self.token_url, json=self.unscoped_token, headers=headers) headers = {'X-Subject-Token': self.scoped_token_id} auth_url = self.AUTH_URL + '/auth/tokens' self.scoped_mock = self.requests_mock.post(auth_url, json=self.scoped_token, headers=headers) def get_plugin(self, **kwargs): kwargs.setdefault('auth_url', self.AUTH_URL) kwargs.setdefault('protocol', self.protocol) kwargs.setdefault('identity_provider', self.idp) return TesterFederationPlugin(**kwargs) def test_federated_url(self): plugin = self.get_plugin() self.assertEqual(self.token_url, plugin.federated_token_url) def test_unscoped_behaviour(self): sess = session.Session(auth=self.get_plugin()) self.assertEqual(self.unscoped_token_id, sess.get_token()) self.assertTrue(self.unscoped_mock.called) self.assertFalse(self.scoped_mock.called) def test_scoped_behaviour(self): auth = self.get_plugin(project_id=self.scoped_token.project_id) sess = session.Session(auth=auth) self.assertEqual(self.scoped_token_id, sess.get_token()) self.assertTrue(self.unscoped_mock.called) self.assertTrue(self.scoped_mock.called) def test_options(self): opts = [o.name for o in v3.FederationBaseAuth.get_options()] self.assertIn('protocol', opts) self.assertIn('identity-provider', opts) class K2KAuthPluginTest(utils.TestCase): TEST_ROOT_URL = 'http://127.0.0.1:5000/' TEST_URL = '%s%s' % (TEST_ROOT_URL, 'v3') TEST_PASS = 'password' REQUEST_ECP_URL = TEST_URL + '/auth/OS-FEDERATION/saml2/ecp' SP_ROOT_URL = 'https://sp1.com/v3' SP_ID = 'sp1' SP_URL = 'https://sp1.com/Shibboleth.sso/SAML2/ECP' SP_AUTH_URL = (SP_ROOT_URL + '/OS-FEDERATION/identity_providers' '/testidp/protocols/saml2/auth') SERVICE_PROVIDER_DICT = { 'id': SP_ID, 'auth_url': SP_AUTH_URL, 'sp_url': SP_URL } def setUp(self): super(K2KAuthPluginTest, self).setUp() self.token_v3 = fixture.V3Token() self.token_v3.add_service_provider( self.SP_ID, self.SP_AUTH_URL, self.SP_URL) self.session = session.Session() self.k2kplugin = self.get_plugin() def _get_base_plugin(self): self.stub_url('POST', ['auth', 'tokens'], headers={'X-Subject-Token': uuid.uuid4().hex}, json=self.token_v3) return v3.Password(self.TEST_URL, username=self.TEST_USER, password=self.TEST_PASS) def _mock_k2k_flow_urls(self): # The IdP should return a ECP wrapped assertion when requested self.requests_mock.register_uri( 'POST', self.REQUEST_ECP_URL, content=six.b(k2k_fixtures.ECP_ENVELOPE), headers={'Content-Type': 'application/vnd.paos+xml'}, status_code=200) # The SP should respond with a 302 self.requests_mock.register_uri( 'POST', self.SP_URL, content=six.b(k2k_fixtures.TOKEN_BASED_ECP), headers={'Content-Type': 'application/vnd.paos+xml'}, status_code=302) # Should not follow the redirect URL, but use the auth_url attribute self.requests_mock.register_uri( 'GET', self.SP_AUTH_URL, json=k2k_fixtures.UNSCOPED_TOKEN, headers={'X-Subject-Token': k2k_fixtures.UNSCOPED_TOKEN_HEADER}) def get_plugin(self, **kwargs): kwargs.setdefault('base_plugin', self._get_base_plugin()) kwargs.setdefault('service_provider', self.SP_ID) return v3.Keystone2Keystone(**kwargs) def test_options(self): opts = [o.name for o in v3.Keystone2Keystone.get_options()] self.assertIn('service-provider', opts) def test_remote_url(self): remote_auth_url = self.k2kplugin._remote_auth_url(self.SP_AUTH_URL) self.assertEqual(self.SP_ROOT_URL, remote_auth_url) def test_fail_getting_ecp_assertion(self): self.requests_mock.register_uri( 'POST', self.REQUEST_ECP_URL, status_code=401) self.assertRaises(exceptions.AuthorizationFailure, self.k2kplugin._get_ecp_assertion, self.session) def test_get_ecp_assertion_empty_response(self): self.requests_mock.register_uri( 'POST', self.REQUEST_ECP_URL, headers={'Content-Type': 'application/vnd.paos+xml'}, content=six.b(''), status_code=200) self.assertRaises(exceptions.InvalidResponse, self.k2kplugin._get_ecp_assertion, self.session) def test_get_ecp_assertion_wrong_headers(self): self.requests_mock.register_uri( 'POST', self.REQUEST_ECP_URL, headers={'Content-Type': uuid.uuid4().hex}, content=six.b(''), status_code=200) self.assertRaises(exceptions.InvalidResponse, self.k2kplugin._get_ecp_assertion, self.session) def test_send_ecp_authn_response(self): self._mock_k2k_flow_urls() # Perform the request response = self.k2kplugin._send_service_provider_ecp_authn_response( self.session, self.SP_URL, self.SP_AUTH_URL) # Check the response self.assertEqual(k2k_fixtures.UNSCOPED_TOKEN_HEADER, response.headers['X-Subject-Token']) def test_end_to_end_workflow(self): self._mock_k2k_flow_urls() auth_ref = self.k2kplugin.get_auth_ref(self.session) self.assertEqual(k2k_fixtures.UNSCOPED_TOKEN_HEADER, auth_ref.auth_token)