Merge pull request #336 from dinkom/stormpath-325

Saml IdP url builder

Author: rdegges

stormpath/saml/__init__.py

@@ -0,0 +1 @@
+from .saml_idp_url_builder import SamlIdpUrlBuilder

stormpath/saml/saml_idp_url_builder.py

@@ -0,0 +1,58 @@
+try:
+    from urllib import urlencode
+except ImportError:
+    from urllib.parse import urlencode
+from uuid import uuid4
+from datetime import datetime
+import jwt
+from oauthlib.common import to_unicode
+
+
+class SamlIdpUrlBuilder(object):
+
+    def __init__(self, application):
+        self.application = application
+
+    def _get_service_provider(self):
+        return self.application.saml_policy.service_provider
+
+    def build(self, options=None):
+        service_provider = self._get_service_provider()
+        api_key_secret = self.application._client.auth.secret
+        api_key_id = self.application._client.auth.id
+
+        try:
+            jti = uuid4().get_hex()
+        except AttributeError:
+            jti = uuid4().hex
+
+        claims = {
+            'iat': datetime.utcnow(),
+            'jti': jti,
+            'iss': api_key_id
+        }
+
+        if options:
+            if 'cb_uri' in options:
+                claims['cb_uri'] = options['cb_uri']
+
+            if 'ash' in options:
+                claims['ash'] = options['ash']
+
+            if 'onk' in options:
+                claims['onk'] = options['onk']
+
+            if 'state' in options:
+                claims['state'] = options['state']
+
+        jwt_signature = to_unicode(
+            jwt.encode(
+                claims, api_key_secret, 'HS256', headers={'kid': api_key_id}),
+            'UTF-8')
+        url_params = {'accessToken': jwt_signature}
+        sso_initiation_endpoint = service_provider.sso_initiation_endpoint.href
+
+        encoded_params = urlencode(url_params)
+        init_url = "%s?%s" % (sso_initiation_endpoint, encoded_params)
+
+        return init_url

tests/live/test_saml.py

@@ -0,0 +1,54 @@
+import jwt
+from stormpath.saml import SamlIdpUrlBuilder
+from tests.live.base import AuthenticatedLiveBase
+
+
+class TestSamlIdpUrlBuilder(AuthenticatedLiveBase):
+
+    def setUp(self):
+        super(TestSamlIdpUrlBuilder, self).setUp()
+
+        self.app_name = self.get_random_name()
+        self.app = self.client.applications.create({
+            'name': self.app_name,
+            'description': 'test app'
+        })
+
+    def test_build_default_url(self):
+        saml_idp_url_builder = SamlIdpUrlBuilder(self.app)
+        url = saml_idp_url_builder.build()
+
+        self.assertTrue('accessToken' in url)
+
+        token = url.split('accessToken=')[1]
+        result = jwt.decode(token, self.app._client.auth.secret)
+
+        self.assertTrue('iss' in result.keys())
+        self.assertTrue('iat' in result.keys())
+        self.assertTrue('jti' in result.keys())
+
+    def test_build_uri_with_options(self):
+        options = {
+            'cb_uri': 'http://some_cb_uri.com/',
+            'ash': 'ash',
+            'onk': 'onk',
+            'state': 'state'
+        }
+
+        saml_idp_url_builder = SamlIdpUrlBuilder(self.app)
+        url = saml_idp_url_builder.build(options)
+
+        self.assertTrue('accessToken' in url)
+
+        token = url.split('accessToken=')[1]
+        result = jwt.decode(token, self.app._client.auth.secret)
+
+        self.assertTrue('cb_uri' in result.keys())
+        self.assertTrue('ash' in result.keys())
+        self.assertTrue('onk' in result.keys())
+        self.assertTrue('state' in result.keys())
+
+        self.assertEqual(result['cb_uri'], options['cb_uri'])
+        self.assertEqual(result['ash'], options['ash'])
+        self.assertEqual(result['onk'], options['onk'])
+        self.assertEqual(result['state'], options['state'])