Handle fragment encoded auth responses.

Serve a simple HTML page with Javascript manually POSTing the raw auth response from the URL fragment.
zamzterz · May 24, 2019 · f9a5fe0 · consideRatio · Dec 5, 2019 · f9a5fe0
1 parent 3ed2274
commit f9a5fe0
Show file tree

Hide file tree

Showing 6 changed files with 158 additions and 9 deletions.
diff --git a/setup.py b/setup.py
@@ -3,7 +3,7 @@
 setup(
         name='Flask-pyoidc',
         version='2.0.0',
-        packages=find_packages('src'),
+        packages=['flask_pyoidc'],
         package_dir={'': 'src'},
         url='https://github.com/zamzterz/flask-pyoidc',
         license='Apache 2.0',
@@ -14,5 +14,6 @@
             'oic==0.12',
             'Flask',
             'requests'
-        ]
+        ],
+        package_data={'flask_pyoidc': ['files/parse_fragment.html']},
 )
diff --git a/src/flask_pyoidc/auth_response_handler.py b/src/flask_pyoidc/auth_response_handler.py
@@ -88,3 +88,17 @@ def process_auth_response(self, auth_response, expected_state, expected_nonce=No
             raise AuthResponseMismatchingSubjectError('The \'sub\' of userinfo does not match \'sub\' of ID Token.')
 
         return AuthenticationResult(access_token, id_token_claims, id_token_jwt, userinfo_claims)
+
+    @classmethod
+    def expect_fragment_encoded_response(cls, auth_request):
+        if 'response_mode' in auth_request:
+            return auth_request['response_mode'] == 'fragment'
+
+        response_type = set(auth_request['response_type'].split(' '))
+        is_implicit_flow = response_type == {'id_token'} or \
+                           response_type == {'id_token', 'token'}
+        is_hybrid_flow = response_type == {'code', 'id_token'} or \
+                         response_type == {'code', 'token'} or \
+                         response_type == {'code', 'id_token', 'token'}
+
+        return is_implicit_flow or is_hybrid_flow
diff --git a/src/flask_pyoidc/files/parse_fragment.html b/src/flask_pyoidc/files/parse_fragment.html
@@ -0,0 +1,17 @@
+<html>
+<script>
+    var req = new XMLHttpRequest();
+    // using POST so query isn't logged
+    req.open('POST', '/redirect_uri', true);
+    req.setRequestHeader('Content-Type', 'application/x-www-form-urlencoded');
+
+    req.onreadystatechange = function () {
+        if (req.readyState === this.DONE) {
+            var location = req.responseText;
+            console.log(location);
+            window.location = location;
+        }
+    };
+    req.send(location.hash.substring(1));
+</script>
+</html>
diff --git a/src/flask_pyoidc/flask_pyoidc.py b/src/flask_pyoidc/flask_pyoidc.py
@@ -18,6 +18,7 @@
 import functools
 import json
 import logging
+import pkg_resources
 from flask import current_app
 from flask.helpers import url_for
 from oic import rndstr
@@ -30,6 +31,10 @@
 
 logger = logging.getLogger(__name__)
 
+try:
+    from urllib.parse import parse_qsl
+except ImportError:
+    from urlparse import parse_qsl
 
 class OIDCAuthentication(object):
     """
@@ -56,7 +61,10 @@ def __init__(self, provider_configurations, app=None):
 
     def init_app(self, app):
         # setup redirect_uri as a flask route
-        app.add_url_rule('/redirect_uri', self.REDIRECT_URI_ENDPOINT, self._handle_authentication_response)
+        app.add_url_rule('/redirect_uri',
+                         self.REDIRECT_URI_ENDPOINT,
+                         self._handle_authentication_response,
+                         methods=['GET', 'POST'])
 
         # dynamically add the Flask redirect uri to the client info
         with app.app_context():
@@ -95,22 +103,42 @@ def _authenticate(self, client, interactive=True):
         login_url = client.authentication_request(flask.session['state'],
                                                   flask.session['nonce'],
                                                   extra_auth_params)
+
+        auth_params = dict(parse_qsl(login_url.split('?')[1]))
+        flask.session['fragment_encoded_response'] = AuthResponseHandler.expect_fragment_encoded_response(auth_params)
         return redirect(login_url)
 
     def _handle_authentication_response(self):
+        has_error = flask.request.args.get('error', False, lambda x: bool(int(x)))
+        if has_error:
+            if 'error' in flask.session:
+                return self._show_error_response(flask.session.pop('error'))
+            return 'Something went wrong.'
+
+        if flask.session.pop('fragment_encoded_response', False):
+            return pkg_resources.resource_string(__name__, 'files/parse_fragment.html').decode('utf-8')
+
+        is_processing_fragment_encoded_response = flask.request.method == 'POST'
+
+        if is_processing_fragment_encoded_response:
+            auth_resp = flask.request.form
+        else:
+            auth_resp = flask.request.args
+
         client = self.clients[UserSession(flask.session).current_provider]
 
-        authn_resp = client.parse_authentication_response(flask.request.args)
+        authn_resp = client.parse_authentication_response(auth_resp)
         logger.debug('received authentication response: %s', authn_resp.to_json())
 
         try:
             result = AuthResponseHandler(client).process_auth_response(authn_resp,
                                                                        flask.session.pop('state'),
                                                                        flask.session.pop('nonce'))
         except AuthResponseErrorResponseError as e:
-            return self._handle_error_response(e.error_response)
+            return self._handle_error_response(e.error_response, is_processing_fragment_encoded_response)
         except AuthResponseProcessError as e:
-            return self._handle_error_response({'error': 'unexpected_error', 'error_description': str(e)})
+            return self._handle_error_response({'error': 'unexpected_error', 'error_description': str(e)},
+                                               is_processing_fragment_encoded_response)
 
         if current_app.config.get('OIDC_SESSION_PERMANENT', True):
             flask.session.permanent = True
@@ -121,9 +149,22 @@ def _handle_authentication_response(self):
                                           result.userinfo_claims)
 
         destination = flask.session.pop('destination')
+        if is_processing_fragment_encoded_response:
+            # if the POST request was from the JS page handling fragment encoded responses we need to return
+            # the destination URL as the response body
+            return destination
+
         return redirect(destination)
 
-    def _handle_error_response(self, error_response):
+    def _handle_error_response(self, error_response, should_redirect=False):
+        if should_redirect:
+            # if the current request was from the JS page handling fragment encoded responses we need to return
+            # a URL for the error page to redirect to
+            flask.session['error'] = error_response
+            return '/redirect_uri?error=1'
+        return self._show_error_response(error_response)
+
+    def _show_error_response(self, error_response):
         logger.error(json.dumps(error_response))
         if self._error_view:
             error = {k: error_response[k] for k in ['error', 'error_description'] if k in error_response}

diff --git a/tests/test_auth_response_handler.py b/tests/test_auth_response_handler.py
@@ -84,3 +84,26 @@ def test_should_handle_token_response_without_id_token(self, client_mock):
                                                                         self.TOKEN_RESPONSE['id_token']['nonce'])
         assert result.access_token == 'test_token'
         assert result.id_token_claims is None
+
+    @pytest.mark.parametrize('response_type, expected', [
+        ('code', False),  # Authorization Code Flow
+        ('id_token', True),  # Implicit Flow
+        ('id_token token', True),  # Implicit Flow
+        ('code id_token', True),  # Hybrid Flow
+        ('code token', True),  # Hybrid Flow
+        ('code id_token token', True)  # Hybrid Flow
+    ])
+    def test_expect_fragment_encoded_response_by_response_type(self, response_type, expected):
+        assert AuthResponseHandler.expect_fragment_encoded_response({'response_type': response_type}) is expected
+
+    @pytest.mark.parametrize('response_type, response_mode, expected', [
+        ('code', 'fragment', True),
+        ('id_token', 'query', False),
+        ('code token', 'form_post', False),
+    ])
+    def test_expect_fragment_encoded_response_with_non_default_response_mode(self,
+                                                                             response_type,
+                                                                             response_mode,
+                                                                             expected):
+        auth_req = {'response_type': response_type, 'response_mode': response_mode}
+        assert AuthResponseHandler.expect_fragment_encoded_response(auth_req) is expected
diff --git a/tests/test_flask_pyoidc.py b/tests/test_flask_pyoidc.py
@@ -95,7 +95,7 @@ def test_reauthenticate_silent_if_session_expired(self):
         with self.app.test_request_context('/'):
             now = time.time()
             with patch('time.time') as time_mock:
-                time_mock.return_value = now - 1 # authenticated in the past
+                time_mock.return_value = now - 1  # authenticated in the past
                 UserSession(flask.session, self.PROVIDER_NAME).update()
             auth_redirect = authn.oidc_auth(self.PROVIDER_NAME)(view_mock)()
 
@@ -111,6 +111,18 @@ def test_dont_reauthenticate_silent_if_session_not_expired(self):
             result = authn.oidc_auth(self.PROVIDER_NAME)(view_mock)()
         self.assert_view_mock(view_mock, result)
 
+    @pytest.mark.parametrize('response_type,expected', [
+        ('code', False),
+        ('id_token token', True)
+    ])
+    def test_expected_auth_response_mode_is_set(self, response_type, expected):
+        authn = self.init_app(auth_request_params={'response_type': response_type})
+        view_mock = self.get_view_mock()
+        with self.app.test_request_context('/'):
+            auth_redirect = authn.oidc_auth(self.PROVIDER_NAME)(view_mock)()
+            assert flask.session['fragment_encoded_response'] is expected
+        self.assert_auth_redirect(auth_redirect)
+
     @responses.activate
     def test_should_register_client_if_not_registered_before(self):
         registration_endpoint = self.PROVIDER_BASEURL + '/register'
@@ -229,7 +241,8 @@ def test_handle_implicit_authentication_response(self, time_mock, utc_time_sans_
 
         authn = self.init_app(provider_metadata_extras={'userinfo_endpoint': userinfo_endpoint})
         state = 'test_state'
-        auth_response = AuthorizationResponse(**{'state': state, 'access_token': access_token, 'id_token': id_token_jwt})
+        auth_response = AuthorizationResponse(
+            **{'state': state, 'access_token': access_token, 'id_token': id_token_jwt})
         with self.app.test_request_context('/redirect_uri?{}'.format(auth_response.to_urlencoded())):
             UserSession(flask.session, self.PROVIDER_NAME)
             flask.session['destination'] = '/'
@@ -242,6 +255,46 @@ def test_handle_implicit_authentication_response(self, time_mock, utc_time_sans_
             assert session.id_token_jwt == id_token_jwt
             assert session.userinfo == userinfo
 
+    def test_handle_authentication_response_POST(self):
+        access_token = 'test_access_token'
+        state = 'test_state'
+
+        authn = self.init_app()
+        auth_response = AuthorizationResponse(**{'state': state, 'access_token': access_token})
+
+        with self.app.test_request_context('/redirect_uri',
+                                           method='POST',
+                                           data=auth_response.to_dict(),
+                                           mimetype='application/x-www-form-urlencoded'):
+            UserSession(flask.session, self.PROVIDER_NAME)
+            flask.session['destination'] = '/test'
+            flask.session['state'] = state
+            flask.session['nonce'] = 'test_nonce'
+            response = authn._handle_authentication_response()
+            session = UserSession(flask.session)
+            assert session.access_token == access_token
+            assert response == '/test'
+
+    def test_handle_authentication_response_fragment_encoded(self):
+        authn = self.init_app()
+        with self.app.test_request_context('/redirect_uri'):
+            flask.session['fragment_encoded_response'] = True
+            response = authn._handle_authentication_response()
+        assert response.startswith('<html>')
+
+    def test_handle_authentication_response_error_message(self):
+        authn = self.init_app()
+        with self.app.test_request_context('/redirect_uri?error=1'):
+            flask.session['error'] = {'error': 'test'}
+            response = authn._handle_authentication_response()
+        assert response == 'Something went wrong with the authentication, please try to login again.'
+
+    def test_handle_authentication_response_error_message_without_stored_error(self):
+        authn = self.init_app()
+        with self.app.test_request_context('/redirect_uri?error=1'):
+            response = authn._handle_authentication_response()
+        assert response == 'Something went wrong.'
+
     @patch('time.time')
     @patch('oic.utils.time_util.utc_time_sans_frac')  # used internally by pyoidc when verifying ID Token
     @responses.activate