Handle OAuth errors and add the possibility to specify an error view.

README.md

@@ -38,3 +38,22 @@ See the [Flask documentation](http://flask.pocoo.org/docs/0.11/config/#builtin-c
 ## Example
 
 Have a look at the example Flask app in [app.py](example/app.py) for an idea of how to use it.
+
+### Specify the error view
+If an OAuth error response is received, either in the authentication or token response, it will be passed along to the
+specified error view. An error view is specified by using the `error_view` decorator:
+
+```python
+from flask import jsonify
+
+@auth.error_view
+def error(error=None, error_description=None):
+    return jsonify({'error': error, 'message': error_description})
+```
+
+The function specified as the error view MUST accept two parameters, `error` and `error_description`, which corresponds
+to the [OIDC/OAuth error parameters](http://openid.net/specs/openid-connect-core-1_0.html#AuthError).
+
+If no error view is specified a generic error message will be displayed to the user.
+
+

example/app.py

@@ -24,5 +24,10 @@ def logout():
     return 'You\'ve been successfully logged out!'
 
 
+@auth.error_view
+def error(error=None, error_description=None):
+    return jsonify({'error': error, 'message': error_description})
+
+
 if __name__ == '__main__':
     app.run(port=PORT)

src/flask_pyoidc/flask_pyoidc.py

@@ -41,6 +41,7 @@ def __init__(self, flask_app, client_registration_info=None, issuer=None,
             self.client.store_registration_info(RegistrationRequest(**client_registration_info))
 
         self.logout_view = None
+        self._error_view = None
 
     def _authenticate(self):
         if 'client_id' not in self.client_registration_info:
@@ -79,6 +80,9 @@ def _handle_authentication_response(self):
         if authn_resp['state'] != flask.session.pop('state'):
             raise ValueError('The \'state\' parameter does not match.')
 
+        if 'error' in authn_resp:
+            return self._handle_error_response(authn_resp)
+
         # do token request
         args = {
             'code': authn_resp['code'],
@@ -88,6 +92,9 @@ def _handle_authentication_response(self):
                                                          request_args=args,
                                                          authn_method=self.client.registration_response.get(
                                                              'token_endpoint_auth_method', 'client_secret_basic'))
+        if 'error' in token_resp:
+            return self._handle_error_response(token_resp)
+
         flask.session['access_token'] = token_resp['access_token']
 
         id_token = None
@@ -116,6 +123,13 @@ def _do_userinfo_request(self, state, userinfo_endpoint_method):
 
         return self.client.do_user_info_request(method=userinfo_endpoint_method, state=state)
 
+    def _handle_error_response(self, error_response):
+        if self._error_view:
+            error = {k: error_response[k] for k in ['error', 'error_description'] if k in error_response}
+            return self._error_view(**error)
+
+        return 'Something went wrong with the authentication, please try to login again.'
+
     def _reauthentication_necessary(self, access_token):
         return not access_token
 
@@ -164,3 +178,7 @@ def wrapper(*args, **kwargs):
             return view_func(*args, **kwargs)
 
         return wrapper
+
+    def error_view(self, view_func):
+        self._error_view = view_func
+        return view_func

tests/test_flask_pyoidc.py

@@ -7,7 +7,7 @@
 from flask import Flask
 from mock import MagicMock
 from oic.oic.message import IdToken, OpenIDSchema
-from six.moves.urllib.parse import parse_qsl, urlparse
+from six.moves.urllib.parse import parse_qsl, urlparse, urlencode
 
 from flask_pyoidc.flask_pyoidc import OIDCAuthentication
 
@@ -197,3 +197,63 @@ def test_oidc_logout_handles_redirects_from_provider(self):
             authn.oidc_logout(callback_mock)()
             assert 'end_session_state' not in flask.session
         assert callback_mock.called
+
+    def test_authentication_error_reponse_calls_to_error_view_if_set(self):
+        state = 'test_tate'
+        error_response = {'error': 'invalid_request', 'error_description': 'test error'}
+        authn = OIDCAuthentication(self.app, provider_configuration_info={'issuer': ISSUER},
+                                   client_registration_info=dict(client_id='abc', client_secret='foo'))
+        error_view_mock = MagicMock()
+        authn._error_view = error_view_mock
+        with self.app.test_request_context('/redirect_uri?{error}&state={state}'.format(
+                error=urlencode(error_response), state=state)):
+            flask.session['state'] = state
+            authn._handle_authentication_response()
+        error_view_mock.assert_called_with(**error_response)
+
+    def test_authentication_error_reponse_returns_default_error_if_no_error_view_set(self):
+        state = 'test_tate'
+        error_response = {'error': 'invalid_request', 'error_description': 'test error'}
+        authn = OIDCAuthentication(self.app, provider_configuration_info={'issuer': ISSUER},
+                                   client_registration_info=dict(client_id='abc', client_secret='foo'))
+        with self.app.test_request_context('/redirect_uri?{error}&state={state}'.format(
+                error=urlencode(error_response), state=state)):
+            flask.session['state'] = state
+            response = authn._handle_authentication_response()
+        assert response == 'Something went wrong with the authentication, please try to login again.'
+
+    @responses.activate
+    def test_token_error_reponse_calls_to_error_view_if_set(self):
+        token_endpoint = ISSUER + '/token'
+        error_response = {'error': 'invalid_request', 'error_description': 'test error'}
+        responses.add(responses.POST, token_endpoint,
+                      body=json.dumps(error_response),
+                      content_type='application/json')
+
+        authn = OIDCAuthentication(self.app, provider_configuration_info={'issuer': ISSUER,
+                                                                          'token_endpoint': token_endpoint},
+                                   client_registration_info=dict(client_id='abc', client_secret='foo'))
+        error_view_mock = MagicMock()
+        authn._error_view = error_view_mock
+        state = 'test_tate'
+        with self.app.test_request_context('/redirect_uri?code=foo&state=' + state):
+            flask.session['state'] = state
+            authn._handle_authentication_response()
+        error_view_mock.assert_called_with(**error_response)
+
+    @responses.activate
+    def test_token_error_reponse_returns_default_error_if_no_error_view_set(self):
+        token_endpoint = ISSUER + '/token'
+        error_response = {'error': 'invalid_request', 'error_description': 'test error'}
+        responses.add(responses.POST, token_endpoint,
+                      body=json.dumps(error_response),
+                      content_type='application/json')
+
+        authn = OIDCAuthentication(self.app, provider_configuration_info={'issuer': ISSUER,
+                                                                          'token_endpoint': token_endpoint},
+                                   client_registration_info=dict(client_id='abc', client_secret='foo'))
+        state = 'test_tate'
+        with self.app.test_request_context('/redirect_uri?code=foo&state=' + state):
+            flask.session['state'] = state
+            response = authn._handle_authentication_response()
+        assert response == 'Something went wrong with the authentication, please try to login again.'