f147a1b095
There is a fair bit of code here. Split the ADFS and SAML2 plugins into their own files so that they are easier to refactor. Change-Id: I76b0d6e7a0dd54d09ef8ed1633e9c85924a9228c
250 lines
9.9 KiB
Python
250 lines
9.9 KiB
Python
# 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 uuid
|
|
|
|
from lxml import etree
|
|
from six.moves import urllib
|
|
|
|
from keystoneauth1 import exceptions
|
|
from keystoneauth1.extras import _saml2 as saml2
|
|
from keystoneauth1.tests.unit import client_fixtures
|
|
from keystoneauth1.tests.unit.extras.saml2 import fixtures as saml2_fixtures
|
|
from keystoneauth1.tests.unit.extras.saml2 import utils
|
|
|
|
|
|
class AuthenticateviaADFSTests(utils.TestCase):
|
|
|
|
GROUP = 'auth'
|
|
|
|
NAMESPACES = {
|
|
's': 'http://www.w3.org/2003/05/soap-envelope',
|
|
'trust': 'http://docs.oasis-open.org/ws-sx/ws-trust/200512',
|
|
'wsa': 'http://www.w3.org/2005/08/addressing',
|
|
'wsp': 'http://schemas.xmlsoap.org/ws/2004/09/policy',
|
|
'a': 'http://www.w3.org/2005/08/addressing',
|
|
'o': ('http://docs.oasis-open.org/wss/2004/01/oasis'
|
|
'-200401-wss-wssecurity-secext-1.0.xsd')
|
|
}
|
|
|
|
USER_XPATH = ('/s:Envelope/s:Header'
|
|
'/o:Security'
|
|
'/o:UsernameToken'
|
|
'/o:Username')
|
|
PASSWORD_XPATH = ('/s:Envelope/s:Header'
|
|
'/o:Security'
|
|
'/o:UsernameToken'
|
|
'/o:Password')
|
|
ADDRESS_XPATH = ('/s:Envelope/s:Body'
|
|
'/trust:RequestSecurityToken'
|
|
'/wsp:AppliesTo/wsa:EndpointReference'
|
|
'/wsa:Address')
|
|
TO_XPATH = ('/s:Envelope/s:Header'
|
|
'/a:To')
|
|
|
|
TEST_TOKEN = uuid.uuid4().hex
|
|
|
|
PROTOCOL = 'saml2'
|
|
|
|
@property
|
|
def _uuid4(self):
|
|
return '4b911420-4982-4009-8afc-5c596cd487f5'
|
|
|
|
def setUp(self):
|
|
super(AuthenticateviaADFSTests, self).setUp()
|
|
|
|
self.IDENTITY_PROVIDER = 'adfs'
|
|
self.IDENTITY_PROVIDER_URL = ('http://adfs.local/adfs/service/trust/13'
|
|
'/usernamemixed')
|
|
self.FEDERATION_AUTH_URL = '%s/%s' % (
|
|
self.TEST_URL,
|
|
'OS-FEDERATION/identity_providers/adfs/protocols/saml2/auth')
|
|
self.SP_ENDPOINT = 'https://openstack4.local/Shibboleth.sso/ADFS'
|
|
|
|
self.adfsplugin = saml2.V3ADFSPassword(
|
|
self.TEST_URL, self.IDENTITY_PROVIDER,
|
|
self.IDENTITY_PROVIDER_URL, self.SP_ENDPOINT,
|
|
self.TEST_USER, self.TEST_TOKEN, self.PROTOCOL)
|
|
|
|
self.ADFS_SECURITY_TOKEN_RESPONSE = utils._load_xml(
|
|
'ADFS_RequestSecurityTokenResponse.xml')
|
|
self.ADFS_FAULT = utils._load_xml('ADFS_fault.xml')
|
|
|
|
def test_get_adfs_security_token(self):
|
|
"""Test ADFSPassword._get_adfs_security_token()."""
|
|
|
|
self.requests_mock.post(
|
|
self.IDENTITY_PROVIDER_URL,
|
|
content=utils.make_oneline(self.ADFS_SECURITY_TOKEN_RESPONSE),
|
|
status_code=200)
|
|
|
|
self.adfsplugin._prepare_adfs_request()
|
|
self.adfsplugin._get_adfs_security_token(self.session)
|
|
|
|
adfs_response = etree.tostring(self.adfsplugin.adfs_token)
|
|
fixture_response = self.ADFS_SECURITY_TOKEN_RESPONSE
|
|
|
|
self.assertEqual(fixture_response, adfs_response)
|
|
|
|
def test_adfs_request_user(self):
|
|
self.adfsplugin._prepare_adfs_request()
|
|
user = self.adfsplugin.prepared_request.xpath(
|
|
self.USER_XPATH, namespaces=self.NAMESPACES)[0]
|
|
self.assertEqual(self.TEST_USER, user.text)
|
|
|
|
def test_adfs_request_password(self):
|
|
self.adfsplugin._prepare_adfs_request()
|
|
password = self.adfsplugin.prepared_request.xpath(
|
|
self.PASSWORD_XPATH, namespaces=self.NAMESPACES)[0]
|
|
self.assertEqual(self.TEST_TOKEN, password.text)
|
|
|
|
def test_adfs_request_to(self):
|
|
self.adfsplugin._prepare_adfs_request()
|
|
to = self.adfsplugin.prepared_request.xpath(
|
|
self.TO_XPATH, namespaces=self.NAMESPACES)[0]
|
|
self.assertEqual(self.IDENTITY_PROVIDER_URL, to.text)
|
|
|
|
def test_prepare_adfs_request_address(self):
|
|
self.adfsplugin._prepare_adfs_request()
|
|
address = self.adfsplugin.prepared_request.xpath(
|
|
self.ADDRESS_XPATH, namespaces=self.NAMESPACES)[0]
|
|
self.assertEqual(self.SP_ENDPOINT, address.text)
|
|
|
|
def test_prepare_sp_request(self):
|
|
assertion = etree.XML(self.ADFS_SECURITY_TOKEN_RESPONSE)
|
|
assertion = assertion.xpath(
|
|
saml2.V3ADFSPassword.ADFS_ASSERTION_XPATH,
|
|
namespaces=saml2.V3ADFSPassword.ADFS_TOKEN_NAMESPACES)
|
|
assertion = assertion[0]
|
|
assertion = etree.tostring(assertion)
|
|
|
|
assertion = assertion.replace(
|
|
b'http://docs.oasis-open.org/ws-sx/ws-trust/200512',
|
|
b'http://schemas.xmlsoap.org/ws/2005/02/trust')
|
|
assertion = urllib.parse.quote(assertion)
|
|
assertion = 'wa=wsignin1.0&wresult=' + assertion
|
|
|
|
self.adfsplugin.adfs_token = etree.XML(
|
|
self.ADFS_SECURITY_TOKEN_RESPONSE)
|
|
self.adfsplugin._prepare_sp_request()
|
|
|
|
self.assertEqual(assertion, self.adfsplugin.encoded_assertion)
|
|
|
|
def test_get_adfs_security_token_authn_fail(self):
|
|
"""Test proper parsing XML fault after bad authentication.
|
|
|
|
An exceptions.AuthorizationFailure should be raised including
|
|
error message from the XML message indicating where was the problem.
|
|
"""
|
|
content = utils.make_oneline(self.ADFS_FAULT)
|
|
self.requests_mock.register_uri('POST',
|
|
self.IDENTITY_PROVIDER_URL,
|
|
content=content,
|
|
status_code=500)
|
|
|
|
self.adfsplugin._prepare_adfs_request()
|
|
self.assertRaises(exceptions.AuthorizationFailure,
|
|
self.adfsplugin._get_adfs_security_token,
|
|
self.session)
|
|
# TODO(marek-denis): Python3 tests complain about missing 'message'
|
|
# attributes
|
|
# self.assertEqual('a:FailedAuthentication', e.message)
|
|
|
|
def test_get_adfs_security_token_bad_response(self):
|
|
"""Test proper handling HTTP 500 and mangled (non XML) response.
|
|
|
|
This should never happen yet, keystoneauth1 should be prepared
|
|
and correctly raise exceptions.InternalServerError once it cannot
|
|
parse XML fault message
|
|
"""
|
|
self.requests_mock.register_uri('POST',
|
|
self.IDENTITY_PROVIDER_URL,
|
|
content=b'NOT XML',
|
|
status_code=500)
|
|
self.adfsplugin._prepare_adfs_request()
|
|
self.assertRaises(exceptions.InternalServerError,
|
|
self.adfsplugin._get_adfs_security_token,
|
|
self.session)
|
|
|
|
# TODO(marek-denis): Need to figure out how to properly send cookies
|
|
# from the request_mock methods.
|
|
def _send_assertion_to_service_provider(self):
|
|
"""Test whether SP issues a cookie."""
|
|
cookie = uuid.uuid4().hex
|
|
|
|
self.requests_mock.post(self.SP_ENDPOINT,
|
|
headers={"set-cookie": cookie},
|
|
status_code=302)
|
|
|
|
self.adfsplugin.adfs_token = self._build_adfs_request()
|
|
self.adfsplugin._prepare_sp_request()
|
|
self.adfsplugin._send_assertion_to_service_provider(self.session)
|
|
|
|
self.assertEqual(1, len(self.session.session.cookies))
|
|
|
|
def test_send_assertion_to_service_provider_bad_status(self):
|
|
self.requests_mock.register_uri('POST', self.SP_ENDPOINT,
|
|
status_code=500)
|
|
|
|
self.adfsplugin.adfs_token = etree.XML(
|
|
self.ADFS_SECURITY_TOKEN_RESPONSE)
|
|
self.adfsplugin._prepare_sp_request()
|
|
|
|
self.assertRaises(
|
|
exceptions.InternalServerError,
|
|
self.adfsplugin._send_assertion_to_service_provider,
|
|
self.session)
|
|
|
|
def test_access_sp_no_cookies_fail(self):
|
|
# clean cookie jar
|
|
self.session.session.cookies = []
|
|
|
|
self.assertRaises(exceptions.AuthorizationFailure,
|
|
self.adfsplugin._access_service_provider,
|
|
self.session)
|
|
|
|
def test_check_valid_token_when_authenticated(self):
|
|
self.requests_mock.register_uri(
|
|
'GET', self.FEDERATION_AUTH_URL,
|
|
json=saml2_fixtures.UNSCOPED_TOKEN,
|
|
headers=client_fixtures.AUTH_RESPONSE_HEADERS)
|
|
|
|
self.session.session.cookies = [object()]
|
|
self.adfsplugin._access_service_provider(self.session)
|
|
response = self.adfsplugin.authenticated_response
|
|
|
|
self.assertEqual(client_fixtures.AUTH_RESPONSE_HEADERS,
|
|
response.headers)
|
|
|
|
self.assertEqual(saml2_fixtures.UNSCOPED_TOKEN['token'],
|
|
response.json()['token'])
|
|
|
|
def test_end_to_end_workflow(self):
|
|
self.requests_mock.register_uri(
|
|
'POST', self.IDENTITY_PROVIDER_URL,
|
|
content=self.ADFS_SECURITY_TOKEN_RESPONSE,
|
|
status_code=200)
|
|
self.requests_mock.register_uri(
|
|
'POST', self.SP_ENDPOINT,
|
|
headers={"set-cookie": 'x'},
|
|
status_code=302)
|
|
self.requests_mock.register_uri(
|
|
'GET', self.FEDERATION_AUTH_URL,
|
|
json=saml2_fixtures.UNSCOPED_TOKEN,
|
|
headers=client_fixtures.AUTH_RESPONSE_HEADERS)
|
|
|
|
# NOTE(marek-denis): We need to mimic this until self.requests_mock can
|
|
# issue cookies properly.
|
|
self.session.session.cookies = [object()]
|
|
token = self.adfsplugin.get_auth_ref(self.session)
|
|
self.assertEqual(client_fixtures.AUTH_SUBJECT_TOKEN, token.auth_token)
|