Add support for RP-Initiated logout.

As described in http://openid.net/specs/openid-connect-session-1_0.html#RPLogout.

example/app.py

@@ -18,5 +18,11 @@ def index():
                    userinfo=flask.g.userinfo.to_dict())
 
 
+@app.route('/logout')
+@auth.oidc_logout
+def logout():
+    return 'You\'ve been successfully logged out!'
+
+
 if __name__ == '__main__':
     app.run(port=PORT)

setup.py

@@ -11,7 +11,7 @@
         author_email='rebecka.gulliksson@umu.se',
         description='Flask extension for OpenID Connect authentication.',
         install_requires=[
-            'oic==0.8.3',
+            'oic==0.9.1.0',
             'Flask'
         ]
 )

src/flask_pyoidc/flask_pyoidc.py

@@ -5,7 +5,7 @@
 from oic import rndstr
 from oic.oic import Client
 from oic.oic.message import ProviderConfigurationResponse, RegistrationRequest, \
-    AuthorizationResponse, IdToken, OpenIDSchema
+    AuthorizationResponse, IdToken, OpenIDSchema, EndSessionRequest
 from oic.utils.authn.client import CLIENT_AUTHN_METHOD
 from werkzeug.utils import redirect
 
@@ -40,12 +40,20 @@ def __init__(self, flask_app, client_registration_info=None, issuer=None,
         if client_registration_info and 'client_id' in client_registration_info:
             # static client info provided
             self.client.store_registration_info(RegistrationRequest(**client_registration_info))
-        else:
+
+        self.logout_view = None
+
+    def _authenticate(self):
+        if 'client_id' not in self.client_registration_info:
             # do dynamic registration
+            if self.logout_view:
+                # handle support for logout
+                with self.app.app_context():
+                    self.client_registration_info['post_logout_redirect_uris'] = [url_for(self.logout_view.__name__,
+                                                                                          _external=True)]
             self.client.register(self.client.provider_info['registration_endpoint'],
                                  **self.client_registration_info)
 
-    def _authenticate(self):
         flask.session['destination'] = flask.request.url
         flask.session['state'] = rndstr()
         flask.session['nonce'] = rndstr()
@@ -93,6 +101,7 @@ def _handle_authentication_response(self):
 
         # store the current user session
         flask.session['id_token'] = id_token.to_dict()
+        flask.session['id_token_jwt'] = id_token.jwt
         flask.session['access_token'] = access_token
         if userinfo:
             flask.session['userinfo'] = userinfo.to_dict()
@@ -127,3 +136,35 @@ def _unpack_user_session(self):
         userinfo_dict = flask.session.pop('userinfo', None)
         if userinfo_dict:
             flask.g.userinfo = OpenIDSchema().from_dict(userinfo_dict)
+
+    def _logout(self):
+        id_token_jwt = flask.session['id_token_jwt']
+        flask.session.clear()
+
+        if 'end_session_endpoint' in self.client.provider_info:
+            flask.session['end_session_state'] = rndstr()
+            end_session_request = EndSessionRequest(
+                id_token_hint=id_token_jwt,
+                post_logout_redirect_uri=self.client_registration_info['post_logout_redirect_uris'][0],
+                state=flask.session['end_session_state'])
+            return redirect(end_session_request.request(self.client.provider_info['end_session_endpoint']), 303)
+
+        return None
+
+    def oidc_logout(self, view_func):
+        self.logout_view = view_func
+
+        @functools.wraps(view_func)
+        def wrapper(*args, **kwargs):
+            if 'state' in flask.request.args:
+                # returning redirect from provider
+                assert flask.request.args['state'] == flask.session.pop('end_session_state')
+                return view_func(*args, **kwargs)
+
+            redirect_to_provider = self._logout()
+            if redirect_to_provider:
+                return redirect_to_provider
+
+            return view_func(*args, **kwargs)
+
+        return wrapper

tests/test_flask_pyoidc.py

@@ -103,3 +103,97 @@ def test_dont_reauthenticate_with_valid_id_token(self):
             authn.oidc_auth(callback_mock)()
         assert not client_mock.construct_AuthorizationRequest.called
         assert callback_mock.called is True
+
+    def test_logout(self):
+        end_session_endpoint = 'https://provider.example.com/end_session'
+        post_logout_uri = 'https://client.example.com/post_logout'
+        authn = OIDCAuthentication(self.app,
+                                   provider_configuration_info={'issuer': ISSUER,
+                                                                'end_session_endpoint': end_session_endpoint},
+                                   client_registration_info={'client_id': 'foo',
+                                                             'post_logout_redirect_uris': [post_logout_uri]})
+        id_token = IdToken(**{'sub': 'sub1', 'nonce': 'nonce'})
+        with self.app.test_request_context('/logout'):
+            flask.session['access_token'] = 'abcde'
+            flask.session['userinfo'] = {'foo': 'bar', 'abc': 'xyz'}
+            flask.session['id_token'] = id_token.to_dict()
+            flask.session['id_token_jwt'] = id_token.to_jwt()
+            end_session_redirect = authn._logout()
+
+            assert all(k not in flask.session for k in ['access_token', 'userinfo', 'id_token', 'id_token_jwt'])
+
+            assert end_session_redirect.status_code == 303
+            assert end_session_redirect.headers['Location'].startswith(end_session_endpoint)
+            parsed_request = dict(parse_qsl(urlparse(end_session_redirect.headers['Location']).query))
+            assert parsed_request['state'] == flask.session['end_session_state']
+            assert parsed_request['id_token_hint'] == id_token.to_jwt()
+            assert parsed_request['post_logout_redirect_uri'] == post_logout_uri
+
+    def test_logout_handles_provider_without_end_session_endpoint(self):
+        post_logout_uri = 'https://client.example.com/post_logout'
+        authn = OIDCAuthentication(self.app,
+                                   provider_configuration_info={'issuer': ISSUER},
+                                   client_registration_info={'client_id': 'foo',
+                                                             'post_logout_redirect_uris': [post_logout_uri]})
+        id_token = IdToken(**{'sub': 'sub1', 'nonce': 'nonce'})
+        with self.app.test_request_context('/logout'):
+            flask.session['access_token'] = 'abcde'
+            flask.session['userinfo'] = {'foo': 'bar', 'abc': 'xyz'}
+            flask.session['id_token'] = id_token.to_dict()
+            flask.session['id_token_jwt'] = id_token.to_jwt()
+            end_session_redirect = authn._logout()
+
+            assert all(k not in flask.session for k in ['access_token', 'userinfo', 'id_token', 'id_token_jwt'])
+            assert end_session_redirect is None
+
+    def test_oidc_logout_redirects_to_provider(self):
+        end_session_endpoint = 'https://provider.example.com/end_session'
+        post_logout_uri = 'https://client.example.com/post_logout'
+        authn = OIDCAuthentication(self.app,
+                                   provider_configuration_info={'issuer': ISSUER,
+                                                                'end_session_endpoint': end_session_endpoint},
+                                   client_registration_info={'client_id': 'foo',
+                                                             'post_logout_redirect_uris': [post_logout_uri]})
+        callback_mock = MagicMock()
+        callback_mock.__name__ = 'test_callback'  # required for Python 2
+        id_token = IdToken(**{'sub': 'sub1', 'nonce': 'nonce'})
+        with self.app.test_request_context('/logout'):
+            flask.session['id_token_jwt'] = id_token.to_jwt()
+            resp = authn.oidc_logout(callback_mock)()
+        assert resp.status_code == 303
+        assert not callback_mock.called
+
+    def test_oidc_logout_redirects_to_provider(self):
+        end_session_endpoint = 'https://provider.example.com/end_session'
+        post_logout_uri = 'https://client.example.com/post_logout'
+        authn = OIDCAuthentication(self.app,
+                                   provider_configuration_info={'issuer': ISSUER,
+                                                                'end_session_endpoint': end_session_endpoint},
+                                   client_registration_info={'client_id': 'foo',
+                                                             'post_logout_redirect_uris': [post_logout_uri]})
+        callback_mock = MagicMock()
+        callback_mock.__name__ = 'test_callback'  # required for Python 2
+        id_token = IdToken(**{'sub': 'sub1', 'nonce': 'nonce'})
+        with self.app.test_request_context('/logout'):
+            flask.session['id_token_jwt'] = id_token.to_jwt()
+            resp = authn.oidc_logout(callback_mock)()
+            assert authn.logout_view == callback_mock
+        assert resp.status_code == 303
+        assert not callback_mock.called
+
+    def test_oidc_logout_handles_redirects_from_provider(self):
+        end_session_endpoint = 'https://provider.example.com/end_session'
+        post_logout_uri = 'https://client.example.com/post_logout'
+        authn = OIDCAuthentication(self.app,
+                                   provider_configuration_info={'issuer': ISSUER,
+                                                                'end_session_endpoint': end_session_endpoint},
+                                   client_registration_info={'client_id': 'foo',
+                                                             'post_logout_redirect_uris': [post_logout_uri]})
+        callback_mock = MagicMock()
+        callback_mock.__name__ = 'test_callback'  # required for Python 2
+        state = 'end_session_123'
+        with self.app.test_request_context('/logout?state=' + state):
+            flask.session['end_session_state'] = state
+            authn.oidc_logout(callback_mock)()
+            assert 'end_session_state' not in flask.session
+        assert callback_mock.called