Merge pull request #8 from its-dirg/user-logout

RebeckaG · web-flow · commit af43d1f1a99b · 2016-09-27T13:47:07.000+02:00
Add support for user logout.
diff --git a/README.md b/README.md
@@ -243,6 +243,29 @@ def request_contains_software_statement(registration_request):
 provider.registration_request_validators.append(request_contains_software_statement)
 ```
 
+# User logout
+
+RP-initiated logout, as described in [Section 5 of OpenID Connect Session Management](http://openid.net/specs/openid-connect-session-1_0.html#RPLogout)
+is supported. The parsed request should be passed to `Provider.logout_user` together with any known subject identifier
+for the user, and then `Provider.do_post_logout_redirect` should be called do obey any valid `post_logout_redirect_uri`
+included in the request.
+
+```python
+def end_session_endpoint(request):
+    end_session_request = EndSessionRequest().deserialize(request.get_data().decode('utf-8'))
+    
+    try:
+        provider.logout_user(session.get('sub'), end_session_request)
+    except InvalidSubjectIdentifier as e:
+        return HTTPResponse('Logout unsuccessful!', content-type='text/html', status=400)
+        
+    redirect_url = provider.do_post_logout_redirect(end_session_request)
+    if redirect_url:
+        return HTTPResponse(redirect_url, status=303)
+
+    return HTTPResponse('Logout successful!', content-type='text/html')
+```
+
 # Exceptions
 All exceptions, except `AuthorizationError`, inherits from `ValueError`. However it might be necessary to distinguish
 between them to send the correct error message back to the client according to the OpenID Connect specifications.
diff --git a/src/pyop/authz_state.py b/src/pyop/authz_state.py
@@ -335,3 +335,13 @@ def get_subject_identifier_for_code(self, authorization_code):
             raise InvalidAuthorizationCode('{} unknown'.format(authorization_code))
 
         return self.authorization_codes[authorization_code]['sub']
+
+    def delete_state_for_subject_identifier(self, subject_identifier):
+        # type (str) -> None
+        if not self._is_valid_subject_identifier(subject_identifier):
+            raise InvalidSubjectIdentifier('Trying to delete state for unknown subject identifier')
+
+        for tokens in [self.authorization_codes, self.access_tokens]:
+            tokens_to_remove = [k for k, v in tokens.items() if v['sub'] == subject_identifier]
+            for ac in tokens_to_remove:
+                tokens.pop(ac, None)
diff --git a/src/pyop/provider.py b/src/pyop/provider.py
@@ -15,6 +15,8 @@
 from oic.oic.message import AccessTokenResponse
 from oic.oic.message import AuthorizationRequest
 from oic.oic.message import AuthorizationResponse
+from oic.oic.message import EndSessionRequest
+from oic.oic.message import EndSessionResponse
 from oic.oic.message import IdToken
 from oic.oic.message import OpenIDSchema
 from oic.oic.message import ProviderConfigurationResponse
@@ -486,3 +488,36 @@ def handle_client_registration_request(self, request, http_headers=None):
         registration_resp = RegistrationResponse(**response_params)
         logger.debug('registration_resp=%s from registration_req', registration_resp, registration_req)
         return registration_resp
+
+    def logout_user(self, subject_identifier=None, end_session_request=None):
+        # type: (Optional[str], Optional[oic.oic.message.EndSessionRequest]) -> None
+        if not end_session_request:
+            end_session_request = EndSessionRequest()
+        if 'id_token_hint' in end_session_request:
+            id_token = IdToken().from_jwt(end_session_request['id_token_hint'], key=[self.signing_key])
+            subject_identifier = id_token['sub']
+
+        self.authz_state.delete_state_for_subject_identifier(subject_identifier)
+
+    def do_post_logout_redirect(self, end_session_request):
+        # type: (oic.oic.message.EndSessionRequest) -> oic.oic.message.EndSessionResponse
+        if 'post_logout_redirect_uri' not in end_session_request:
+            return None
+
+        client_id = None
+        if 'id_token_hint' in end_session_request:
+            id_token = IdToken().from_jwt(end_session_request['id_token_hint'], key=[self.signing_key])
+            client_id = id_token['aud'][0]
+
+        if 'post_logout_redirect_uri' in end_session_request:
+            if not client_id:
+                return None
+            if not end_session_request['post_logout_redirect_uri'] in self.clients[client_id].get(
+                    'post_logout_redirect_uris', []):
+                return None
+
+        end_session_response = EndSessionResponse()
+        if 'state' in end_session_request:
+            end_session_response['state'] = end_session_request['state']
+
+        return end_session_response.request(end_session_request['post_logout_redirect_uri'])
diff --git a/tests/pyop/test_authz_state.py b/tests/pyop/test_authz_state.py
@@ -413,3 +413,21 @@ def test_get_subject_identifier_for_code(self, authorization_state, authorizatio
         authz_code = authorization_state.create_authorization_code(authorization_request, self.TEST_SUBJECT_IDENTIFIER)
         sub = authorization_state.get_subject_identifier_for_code(authz_code)
         assert sub == self.TEST_SUBJECT_IDENTIFIER
+
+    def test_remove_state_for_subject_identifier(self, authorization_state, authorization_request):
+        self.set_valid_subject_identifier(authorization_state)
+        authz_code1 = authorization_state.create_authorization_code(authorization_request, self.TEST_SUBJECT_IDENTIFIER)
+        authz_code2 = authorization_state.create_authorization_code(authorization_request, self.TEST_SUBJECT_IDENTIFIER)
+        access_token1 = authorization_state.create_access_token(authorization_request, self.TEST_SUBJECT_IDENTIFIER)
+        access_token2 = authorization_state.create_access_token(authorization_request, self.TEST_SUBJECT_IDENTIFIER)
+
+        authorization_state.delete_state_for_subject_identifier(self.TEST_SUBJECT_IDENTIFIER)
+
+        for ac in [authz_code1, authz_code2]:
+            assert ac not in authorization_state.authorization_codes
+        for at in [access_token1, access_token2]:
+            assert at.value not in authorization_state.access_tokens
+
+    def test_remove_state_for_unknown_subject_identifier(self, authorization_state):
+        with pytest.raises(InvalidSubjectIdentifier):
+            authorization_state.delete_state_for_subject_identifier('unknown')
diff --git a/tests/pyop/test_provider.py b/tests/pyop/test_provider.py
@@ -12,13 +12,13 @@
 from oic import rndstr
 from oic.oauth2.message import MissingRequiredValue, MissingRequiredAttribute
 from oic.oic import PREFERENCE2PROVIDER
-from oic.oic.message import IdToken, AuthorizationRequest, ClaimsRequest, Claims
+from oic.oic.message import IdToken, AuthorizationRequest, ClaimsRequest, Claims, EndSessionRequest, EndSessionResponse
 
 from pyop.access_token import BearerTokenError
 from pyop.authz_state import AuthorizationState
 from pyop.client_authentication import InvalidClientAuthentication
 from pyop.exceptions import InvalidAuthenticationRequest, AuthorizationError, InvalidTokenRequest, \
-    InvalidClientRegistrationRequest, InvalidAccessToken
+    InvalidClientRegistrationRequest, InvalidAccessToken, InvalidAuthorizationCode, InvalidSubjectIdentifier
 from pyop.provider import Provider, redirect_uri_is_in_registered_redirect_uris, \
     response_type_is_in_registered_response_types
 from pyop.subject_identifier import HashBasedSubjectIdentifierFactory
@@ -69,7 +69,9 @@ def inject_provider(request):
             'redirect_uris': [TEST_REDIRECT_URI],
             'response_types': ['code'],
             'client_secret': TEST_CLIENT_SECRET,
-            'token_endpoint_auth_method': 'client_secret_post'
+            'token_endpoint_auth_method': 'client_secret_post',
+            'post_logout_redirect_uris': ['https://client.example.com/post_logout']
+
         }
     }
 
@@ -558,3 +560,58 @@ class TestProviderJWKS(object):
     def test_jwks(self):
         provider = Provider(rsa_key(), {'issuer': ISSUER}, None, None, None)
         assert provider.jwks == {'keys': [provider.signing_key.serialize()]}
+
+
+@pytest.mark.usefixtures('inject_provider')
+class TestRPInitiatedLogout(object):
+    def test_logout_user_with_subject_identifier(self):
+        auth_req = AuthorizationRequest(response_type='code id_token token', scope='openid', client_id='client1',
+                                        redirect_uri='https://client.example.com/redirect')
+        auth_resp = self.provider.authorize(auth_req, 'user1')
+
+        id_token = IdToken().from_jwt(auth_resp['id_token'], key=[self.provider.signing_key])
+        self.provider.logout_user(subject_identifier=id_token['sub'])
+        with pytest.raises(InvalidAccessToken):
+            self.provider.authz_state.introspect_access_token(auth_resp['access_token'])
+        with pytest.raises(InvalidAuthorizationCode):
+            self.provider.authz_state.exchange_code_for_token(auth_resp['code'])
+
+    def test_logout_user_with_id_token_hint(self):
+        auth_req = AuthorizationRequest(response_type='code id_token token', scope='openid', client_id='client1',
+                                        redirect_uri='https://client.example.com/redirect')
+        auth_resp = self.provider.authorize(auth_req, 'user1')
+
+        self.provider.logout_user(end_session_request=EndSessionRequest(id_token_hint=auth_resp['id_token']))
+        with pytest.raises(InvalidAccessToken):
+            self.provider.authz_state.introspect_access_token(auth_resp['access_token'])
+        with pytest.raises(InvalidAuthorizationCode):
+            self.provider.authz_state.exchange_code_for_token(auth_resp['code'])
+
+    def test_logout_user_with_unknown_subject_identifier(self):
+        with pytest.raises(InvalidSubjectIdentifier):
+            self.provider.logout_user(subject_identifier='unknown')
+
+    def test_post_logout_redirect(self):
+        auth_req = AuthorizationRequest(response_type='code id_token token', scope='openid', client_id='client1',
+                                        redirect_uri='https://client.example.com/redirect')
+        auth_resp = self.provider.authorize(auth_req, 'user1')
+        end_session_request = EndSessionRequest(id_token_hint=auth_resp['id_token'],
+                                                post_logout_redirect_uri='https://client.example.com/post_logout',
+                                                state='state')
+        redirect_url = self.provider.do_post_logout_redirect(end_session_request)
+        assert redirect_url == EndSessionResponse(state='state').request('https://client.example.com/post_logout')
+
+    def test_post_logout_redirect_without_post_logout_redirect_uri(self):
+        assert self.provider.do_post_logout_redirect(EndSessionRequest()) is None
+
+    def test_post_logout_redirect_with_unknown_client_for_post_logout_redirect_uri(self):
+        end_session_request = EndSessionRequest(post_logout_redirect_uri='https://client.example.com/post_logout')
+        assert self.provider.do_post_logout_redirect(end_session_request) is None
+
+    def test_post_logout_redirect_with_unknown_post_logout_redirect_uri(self):
+        auth_req = AuthorizationRequest(response_type='code id_token token', scope='openid', client_id='client1',
+                                        redirect_uri='https://client.example.com/redirect')
+        auth_resp = self.provider.authorize(auth_req, 'user1')
+        end_session_request = EndSessionRequest(id_token_hint=auth_resp['id_token'],
+                                                post_logout_redirect_uri='https://client.example.com/unknown')
+        assert self.provider.do_post_logout_redirect(end_session_request) is None
