Check custom url passed to auth methods

Author: dinkom

stormpath/api_auth.py

@@ -539,6 +539,9 @@ def authenticate(self, id_site_jwt, organization_name_key=None, account_store=No
         """
         if not url:
             url = self.app.href + '/oauth/token'
+        else:
+            if not url.startswith(self.app._client.BASE_URL):
+                raise ValueError('Invalid API url.')
 
         headers = {'Content-Type': 'application/x-www-form-urlencoded'}
         data = {
@@ -597,6 +600,9 @@ def authenticate(self, provider_id, code=None, access_token=None, account_store=
         """
         if not url:
             url = self.app.href + '/oauth/token'
+        else:
+            if not url.startswith(self.app._client.BASE_URL):
+                raise ValueError('Invalid API url.')
 
         headers = {'Content-Type': 'application/x-www-form-urlencoded'}
         data = {
@@ -661,6 +667,9 @@ def authenticate(self, username, password, account_store=None, url=None):
         """
         if not url:
             url = self.app.href + '/oauth/token'
+        else:
+            if not url.startswith(self.app._client.BASE_URL):
+                raise ValueError('Invalid API url.')
 
         headers = {'Content-Type': 'application/x-www-form-urlencoded'}
         data = {
@@ -701,6 +710,9 @@ class ClientCredentialsGrantAuthenticator(Authenticator):
     def authenticate(self, client_id, client_secret, account_store=None, url=None):
         if not url:
             url = self.app.href + '/oauth/token'
+        else:
+            if not url.startswith(self.app._client.BASE_URL):
+                raise ValueError('Invalid API url.')
 
         headers = {'Content-Type': 'application/x-www-form-urlencoded'}
 

tests/live/test_api_auth.py

@@ -1117,6 +1117,50 @@ def test_authenticate_with_account_store_succeeds(self):
         self.assertEqual(result.expires_in, 3600)
         self.assertEqual(result.account.href, self.acc.href)
 
+    def test_with_invalid_url_fails(self):
+        id_site_url = self.app.build_id_site_redirect_url(
+            'https://hi.com')
+
+        resp = get(id_site_url, allow_redirects=False)
+        jwt = resp.headers['Location'].split('jwt=')[1]
+        origin = resp.headers['Location'].split('#')[0]
+
+        resp = get(self.app.href, params={'expand': 'idSiteModel'}, headers={
+            'Authorization': 'Bearer {}'.format(jwt),
+            'Origin': origin,
+            'Referer': origin,
+        })
+
+        next_jwt = resp.headers['Authorization'].split('Bearer ')[1]
+
+        post_url = '%s%s' % (self.app.href, '/loginAttempts')
+        headers = {
+            'Authorization': 'Bearer {}'.format(next_jwt),
+            'Origin': origin,
+            'Referer': origin,
+            'Content Type': 'application/json'
+        }
+
+        user_pass = '%s:%s' % (self.acc.email, self.password)
+        try:
+            encrypted_value = base64.b64encode(bytes(user_pass))
+        except TypeError:
+            # Python 3
+            encrypted_value = base64.b64encode(bytes(user_pass, 'utf-8'))
+            encrypted_value = encrypted_value.decode('utf-8')
+
+        body = {
+            'value': encrypted_value,
+            'type': 'basic'
+        }
+        resp = post(post_url, headers=headers, json=body)
+        final_jwt = resp.headers['stormpath-sso-redirect-location'].split('jwtResponse=')[1]
+
+        authenticator = StormpathTokenGrantAuthenticator(self.app)
+
+        with self.assertRaises(ValueError):
+            authenticator.authenticate(final_jwt, url='http://attack.com')
+
 
 class TestStormpathSocialGrantAuthenticator(ApiKeyBase):
     def setUp(self):
@@ -1251,6 +1295,15 @@ def test_authenticate_with_account_store_succeeds(self):
         self.assertEqual(result.expires_in, 3600)
         self.assertEqual(result.account.email, self.test_user['email'])
 
+    def test_with_invalid_url_fails(self):
+        authenticator = StormpathSocialGrantAuthenticator(self.app)
+        with self.assertRaises(ValueError):
+            authenticator.authenticate(
+                provider_id=Provider.FACEBOOK,
+                access_token=self.test_user['access_token'],
+                url='http://attack.com'
+            )
+
 
 class TestPasswordGrantAuthenticator(ApiKeyBase):
     def setUp(self):
@@ -1378,6 +1431,13 @@ def test_authenticate_with_account_store_org_href_succeeds(self):
         self.assertEqual(result.account.href, self.acc.href)
         self.assertEqual(claims.get('org'), self.org.href)
 
+    def test_with_invalid_url_fails(self):
+        authenticator = PasswordGrantAuthenticator(self.app)
+        with self.assertRaises(ValueError):
+            authenticator.authenticate(self.username,
+                                       self.password,
+                                       url='http://attack.com')
+
 
 class TestClientCredentialsGrantAuthenticator(ApiKeyBase):
 
@@ -1450,6 +1510,13 @@ def test_authenticate_with_account_store_fails(self):
 
         self.assertIsNone(result)
 
+    def test_with_invalid_url_fails(self):
+        authenticator = ClientCredentialsGrantAuthenticator(self.app)
+        with self.assertRaises(ValueError):
+            authenticator.authenticate(self.user_api_key.id,
+                                       self.user_api_key.secret,
+                                       url='http://attack.com')
+
 
 class TestJwtAuthenticator(ApiKeyBase):
     def setUp(self):