contrib: handle login via token

invenio_oauthclient/contrib/cern_openid.py

@@ -82,7 +82,10 @@
     oauth_unlink_external_id
 
 OAUTHCLIENT_CERN_OPENID_REFRESH_TIMEDELTA = timedelta(minutes=-5)
-"""Default interval for refreshing CERN extra data (e.g. groups)."""
+"""Default interval for refreshing CERN extra data (e.g. groups).
+
+False value disabled the refresh.
+"""
 
 OAUTHCLIENT_CERN_OPENID_SESSION_KEY = "identity.cern_openid_provides"
 """Name of session key where CERN roles are stored."""
@@ -95,15 +98,15 @@
     description="Connecting to CERN Organization.",
     icon="",
     logout_url="https://auth.cern.ch/auth/realms/cern/protocol/"
-    "openid-connect/logout",
+               "openid-connect/logout",
     params=dict(
         base_url="https://auth.cern.ch/auth/realms/cern",
         request_token_url=None,
         access_token_url="https://auth.cern.ch/auth/realms/cern/protocol/"
-        "openid-connect/token",
+                         "openid-connect/token",
         access_token_method="POST",
         authorize_url="https://auth.cern.ch/auth/realms/cern/protocol/"
-        "openid-connect/auth",
+                      "openid-connect/auth",
         app_key="CERN_APP_OPENID_CREDENTIALS",
         content_type="application/json",
     ),
@@ -113,9 +116,9 @@
 REMOTE_APP.update(
     dict(
         authorized_handler="invenio_oauthclient.handlers"
-        ":authorized_signup_handler",
+                           ":authorized_signup_handler",
         disconnect_handler="invenio_oauthclient.contrib.cern_openid"
-        ":disconnect_handler",
+                           ":disconnect_handler",
         signup_handler=dict(
             info="invenio_oauthclient.contrib.cern_openid:account_info",
             setup="invenio_oauthclient.contrib.cern_openid:account_setup",
@@ -129,9 +132,9 @@
 REMOTE_REST_APP.update(
     dict(
         authorized_handler="invenio_oauthclient.handlers.rest"
-        ":authorized_signup_handler",
+                           ":authorized_signup_handler",
         disconnect_handler="invenio_oauthclient.contrib.cern_openid"
-        ":disconnect_rest_handler",
+                           ":disconnect_rest_handler",
         signup_handler=dict(
             info="invenio_oauthclient.contrib.cern_openid:account_info_rest",
             setup="invenio_oauthclient.contrib.cern_openid:account_setup",
@@ -148,7 +151,6 @@
 )
 """CERN Openid Remote REST Application."""
 
-
 REMOTE_APP_RESOURCE_API_URL = (
     "https://auth.cern.ch/auth/realms/cern/protocol/openid-connect/userinfo"
 )
@@ -370,26 +372,34 @@ def on_identity_changed(sender, identity):
         disconnect_identity(identity)
         return
 
+    logged_in_via_token = hasattr(current_user, 'login_via_oauth2') \
+        and getattr(current_user, 'login_via_oauth2')
+
     client_id = current_app.config["CERN_APP_OPENID_CREDENTIALS"][
         "consumer_key"
     ]
-    account = RemoteAccount.get(
+    remote_account = RemoteAccount.get(
         user_id=current_user.get_id(), client_id=client_id
     )
     roles = []
-    if account:
-        remote = find_remote_by_client_id(client_id)
-        resource = get_resource(remote)
+
+    if remote_account and not logged_in_via_token:
         refresh = current_app.config.get(
             "OAUTHCLIENT_CERN_OPENID_REFRESH_TIMEDELTA",
             OAUTHCLIENT_CERN_OPENID_REFRESH_TIMEDELTA,
         )
-
-        roles.extend(
-            account_roles_and_extra_data(
-                account, resource, refresh_timedelta=refresh
+        if refresh:
+            remote = find_remote_by_client_id(client_id)
+            resource = get_resource(remote)
+            roles.extend(
+                account_roles_and_extra_data(
+                    remote_account, resource, refresh_timedelta=refresh
+                )
             )
-        )
+        else:
+            roles.extend(remote_account.extra_data["roles"])
+    elif remote_account and logged_in_via_token:
+        roles.extend(remote_account.extra_data["roles"])
 
     extend_identity(identity, roles)
 

tests/test_contrib_cern_openid_rest.py

@@ -11,9 +11,11 @@
 from __future__ import absolute_import
 
 import os
+from datetime import timedelta
 
 import pytest
-from flask import g, session, url_for
+from flask import current_app, g, session, url_for
+from flask_login import current_user
 from flask_security import login_user, logout_user
 from helpers import check_response_redirect_url_args, get_state, \
     mock_remote_get, mock_response
@@ -22,6 +24,7 @@
 from invenio_oauthclient.contrib.cern_openid import \
     OAUTHCLIENT_CERN_OPENID_SESSION_KEY, account_info_rest, \
     disconnect_rest_handler, fetch_extra_data, get_dict_from_response
+from invenio_oauthclient.models import RemoteAccount
 
 from flask_oauthlib.client import OAuthResponse  # noqa isort:skip
 
@@ -196,3 +199,85 @@ def test_account_info_not_allowed_account(app_rest, example_cern_openid_rest):
         "code": 400,
     }
     check_response_redirect_url_args(resp, expected_url_args)
+
+
+def test_identity_changed(app_rest, example_cern_openid_rest, models_fixture):
+    ioc = app_rest.extensions['oauthlib.client']
+
+    # setup the user account via cern_openid
+    with app_rest.test_client() as c:
+
+        # Ensure remote apps have been loaded (due to before first request)
+        resp = c.get(url_for('invenio_oauthclient.rest_login',
+                             remote_app='cern_openid'))
+        assert resp.status_code == 302
+
+        example_response, example_token, example_account_info = \
+            example_cern_openid_rest
+
+        mock_response(app_rest.extensions['oauthlib.client'], 'cern_openid',
+                      example_token)
+        mock_remote_get(ioc, 'cern_openid', example_response)
+
+        resp = c.get(url_for(
+            'invenio_oauthclient.rest_authorized',
+            remote_app='cern_openid', code='test',
+            state=get_state('cern_openid')))
+        assert resp.status_code == 302
+        expected_url_args = {
+            "message": "Successfully authorized.",
+            "code": 200,
+        }
+        check_response_redirect_url_args(resp, expected_url_args)
+
+        assert len(g.identity.provides) == 3
+
+    datastore = app_rest.extensions['invenio-accounts'].datastore
+    user = datastore.find_user(email='john.doe@cern.ch')
+    assert user
+
+    # mark user as logged in via token
+    user.login_via_oauth2 = True
+
+    with app_rest.test_request_context():
+        client_id = current_app.config["CERN_APP_OPENID_CREDENTIALS"][
+            "consumer_key"
+        ]
+        # make sure the roles are cleaned
+        remote_account = RemoteAccount.get(
+            user_id=user.get_id(), client_id=client_id
+        )
+
+        # check if the initial roles are there
+        login_user(user)
+
+        assert current_user.login_via_oauth2
+
+        assert len(g.identity.provides) == 3
+        logout_user()
+
+        remote_account.extra_data.update(roles=[])
+
+        # login the user again
+        login_user(user)
+
+        # check if the cern roles are not fetched from the provider
+        assert len(g.identity.provides) == 2
+        logout_user()
+
+        user.login_via_oauth2 = False
+
+        current_app.config[
+            'OAUTHCLIENT_CERN_OPENID_REFRESH_TIMEDELTA'] = False
+        login_user(user)
+
+        # check that the roles are not refreshed from provider
+        assert len(g.identity.provides) == 2
+        logout_user()
+
+        current_app.config['OAUTHCLIENT_CERN_OPENID_REFRESH_TIMEDELTA'] \
+            = timedelta(microseconds=1)
+        login_user(user)
+
+        # check if roles refreshed from the provider
+        assert len(g.identity.provides) == 3