/usr/lib/python2.7/dist-packages/keystoneauth1/tests/unit/extras/saml2/test_auth_adfs.py is in python-keystoneauth1 3.4.0-0ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 | # 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
from keystoneauth1.tests.unit import matchers
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.SP_ENTITYID = 'https://openstack4.local'
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.assertThat(fixture_response,
matchers.XMLEquals(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_adfs_request_custom_endpointreference(self):
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.SP_ENTITYID)
self.adfsplugin._prepare_adfs_request()
address = self.adfsplugin.prepared_request.xpath(
self.ADDRESS_XPATH, namespaces=self.NAMESPACES)[0]
self.assertEqual(self.SP_ENTITYID, 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)
|