openstack/keystoneauth
Code Issues Proposed changes

Add 303 as redirect code for k2k plugin

Browse Source 
Some service providers, like mod_mellon return a 303 response
upon successful authentication. The "requests" package only
considers 302, as per the following example::

  >>> import requests
  >>> requests.codes['found']
  302

Change-Id: I5797f490f2e57d1c952e769bc0ef4b96c08f9a83
Related-Bug: 1501918
This commit is contained in:
Rodrigo Duarte 2016-04-25 16:19:48 -03:00
parent 9d51a2ae60
commit 1a2a579393
2 changed files with 28 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
keystoneauth1/identity/v3/k2k.py
View File

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import requests
import six
from keystoneauth1 import access
@ -42,6 +41,9 @@ class Keystone2Keystone(federation._Rescoped):
"""Path where the ECP wrapped SAML assertion should be presented to the
Keystone Service Provider."""
HTTP_MOVED_TEMPORARILY = 302
HTTP_SEE_OTHER = 303
def __init__(self, base_plugin, service_provider, **kwargs):
super(Keystone2Keystone, self).__init__(auth_url=None, **kwargs)
@ -147,11 +149,12 @@ class Keystone2Keystone(federation._Rescoped):
authenticated=False,
redirect=False)
# Don't follow HTTP specs - after the HTTP 302 response don't repeat
# the call directed to the Location URL. In this case, this is an
# indication that SAML2 session is now active and protected resource
# Don't follow HTTP specs - after the HTTP 302/303 response don't
# repeat the call directed to the Location URL. In this case, this is
# an indication that SAML2 session is now active and protected resource
# can be accessed.
if response.status_code == requests.codes['found']:
if response.status_code in (self.HTTP_MOVED_TEMPORARILY,
self.HTTP_SEE_OTHER):
response = session.get(
sp_auth_url,
headers={'Content-Type': 'application/vnd.paos+xml'},

24
keystoneauth1/tests/unit/identity/test_identity_v3_federation.py
View File

@ -133,7 +133,7 @@ class K2KAuthPluginTest(utils.TestCase):
username=self.TEST_USER,
password=self.TEST_PASS)
def _mock_k2k_flow_urls(self):
def _mock_k2k_flow_urls(self, redirect_code=302):
# List versions available for auth
self.requests_mock.get(
self.TEST_URL,
@ -148,13 +148,13 @@ class K2KAuthPluginTest(utils.TestCase):
headers={'Content-Type': 'application/vnd.paos+xml'},
status_code=200)
# The SP should respond with a 302
# The SP should respond with a redirect (302 or 303)
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)
status_code=redirect_code)
# Should not follow the redirect URL, but use the auth_url attribute
self.requests_mock.register_uri(
@ -226,12 +226,28 @@ class K2KAuthPluginTest(utils.TestCase):
self.assertEqual(k2k_fixtures.UNSCOPED_TOKEN_HEADER,
response.headers['X-Subject-Token'])
def test_send_ecp_authn_response_303_redirect(self):
self._mock_k2k_flow_urls(redirect_code=303)
# 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)
def test_end_to_end_workflow_303_redirect(self):
self._mock_k2k_flow_urls(redirect_code=303)
auth_ref = self.k2kplugin.get_auth_ref(self.session)
self.assertEqual(k2k_fixtures.UNSCOPED_TOKEN_HEADER,
auth_ref.auth_token)
def test_end_to_end_with_generic_password(self):
# List versions available for auth
self.requests_mock.get(
@ -247,7 +263,7 @@ class K2KAuthPluginTest(utils.TestCase):
headers={'Content-Type': 'application/vnd.paos+xml'},
status_code=200)
# The SP should respond with a 302
# The SP should respond with a redirect (302 or 303)
self.requests_mock.register_uri(
'POST',
self.SP_URL,