Auto-Login to CKAN (but only if the user belongs to ckan-admin group in Keycloak)

ckanext/keycloak/keycloak.py

@@ -21,11 +21,11 @@ def get_token(self, code, redirect_uri):
         return self.get_keycloak_client().token(grant_type="authorization_code", code=code, redirect_uri=redirect_uri)
 
     def get_user_info(self, token):
-        print (token.get('access_token'))
+        # log.info(token.get('access_token'))
         return self.get_keycloak_client().userinfo(token.get('access_token'))
 
     def get_user_groups(self, token):
-        return self.get_keycloak_client().userinfo(token).get('groups', [])
+        return self.get_keycloak_client().userinfo(token.get('access_token')).get('groups', [])
 
     def get_keycloak_admin(self):
         return KeycloakAdmin(

ckanext/keycloak/views.py

@@ -3,7 +3,6 @@
 from ckan.plugins import toolkit as tk
 import ckan.lib.helpers as h
 import ckan.model as model
-from ckan.common import g
 from ckan.views.user import set_repoze_user, RequestResetView
 from ckanext.keycloak.keycloak import KeycloakClient
 import ckanext.keycloak.helpers as helpers
@@ -13,15 +12,16 @@
 
 keycloak = Blueprint('keycloak', __name__, url_prefix='/user')
 
-
 server_url = tk.config.get('ckanext.keycloak.server_url', environ.get('CKANEXT__KEYCLOAK__SERVER_URL'))
 client_id = tk.config.get('ckanext.keycloak.client_id', environ.get('CKANEXT__KEYCLOAK__CLIENT_ID'))
 realm_name = tk.config.get('ckanext.keycloak.realm_name', environ.get('CKANEXT__KEYCLOAK__REALM_NAME'))
 redirect_uri = tk.config.get('ckanext.keycloak.redirect_uri', environ.get('CKANEXT__KEYCLOAK__REDIRECT_URI'))
-client_secret_key = tk.config.get('ckanext.keycloak.client_secret_key', environ.get('CKANEXT__KEYCLOAK__CLIENT_SECRET_KEY'))
+client_secret_key = tk.config.get('ckanext.keycloak.client_secret_key',
+                                  environ.get('CKANEXT__KEYCLOAK__CLIENT_SECRET_KEY'))
 
 client = KeycloakClient(server_url, client_id, realm_name, client_secret_key)
 
+
 def _log_user_into_ckan(resp):
     """ Log the user into different CKAN versions.
     CKAN 2.10 introduces flask-login and login_user method.
@@ -31,33 +31,37 @@ def _log_user_into_ckan(resp):
     """
     if tk.check_ckan_version(min_version="2.10"):
         from ckan.common import login_user
-        login_user(g.user_obj)
+        login_user(tk.g.user_obj)
         return
 
     if tk.check_ckan_version(min_version="2.9.6"):
         user_id = "{},1".format(g.user_obj.id)
     else:
-        user_id = g.user
+        user_id = tk.g.user
     set_repoze_user(user_id, resp)
 
     log.info(u'User {0}<{1}> logged in successfully'.format(g.user_obj.name, g.user_obj.email))
 
+
 def sso():
-    log.info("SSO Login")
-    auth_url = None
     try:
         auth_url = client.get_auth_url(redirect_uri=redirect_uri)
     except Exception as e:
         log.error("Error getting auth url: {}".format(e))
         return tk.abort(500, "Error getting auth url: {}".format(e))
     return tk.redirect_to(auth_url)
 
+
 def sso_login():
     data = tk.request.args
     token = client.get_token(data['code'], redirect_uri)
     userinfo = client.get_user_info(token)
-    log.info("SSO Login: {}".format(userinfo))
-    if userinfo:
+    usergroups = client.get_user_groups(token)
+    log.info("User Info: {}".format(userinfo))
+
+    is_user_ckan_admin = 'ckan_admin' in usergroups
+
+    if userinfo and is_user_ckan_admin:  # only login if admin
         user_dict = {
             'name': helpers.ensure_unique_username_from_email(userinfo['preferred_username']),
             'email': userinfo['email'],
@@ -68,19 +72,23 @@ def sso_login():
             }
         }
         context = {"model": model, "session": model.Session}
-        g.user_obj = helpers.process_user(user_dict)
-        g.user = g.user_obj.name
-        context['user'] = g.user
-        context['auth_user_obj'] = g.user_obj
+        tk.g.user_obj = helpers.process_user(user_dict)
+        tk.g.user = tk.g.user_obj.name
+        context['user'] = tk.g.user
+        context['auth_user_obj'] = tk.g.user_obj
 
         response = tk.redirect_to(tk.url_for('user.me', context))
 
         _log_user_into_ckan(response)
         log.info("Logged in success")
         return response
+    elif userinfo and not is_user_ckan_admin:  # if user exists but not admin
+        # h.flash_error(f'User {userinfo["preferred_username"]} doesn\'t have permission to log in')
+        return tk.redirect_to(tk.url_for('organization.index'))
     else:
         return tk.redirect_to(tk.url_for('user.login'))
 
+
 def reset_password():
     email = tk.request.form.get('user', None)
     if '@' not in email:
@@ -98,9 +106,49 @@ def reset_password():
         return tk.redirect_to(tk.url_for('user.login'))
     return RequestResetView().post()
 
+
+# This endpoint will take care of auto-login
+def sso_autologin():
+    # Check for arguments in the URL to determine the further logic.
+    # If there was a redirect without any arguments, make call to keycloak to check if logged in there.
+    #     Keycloak will return back with 2 options:
+    #       a. if there was a redirect from keycloak with 'code' argument, then complete sso (create user and/or login)
+    #       b. if there was a redirect from keycloak with 'error' argument, it means keycloak login was not done yet
+    data = tk.request.args
+
+    if len(data):
+        if data.get('code'):
+            return tk.redirect_to('keycloak.sso')  # complete login/create user
+        elif data.get('error'):
+            return tk.redirect_to('organization.index')  # redirect to organizations page
+    else:
+        # check keycloak for logged in state
+        try:
+            auth_url = client.get_auth_url(redirect_uri=f"{environ.get('CKAN_SITE_URL')}/user/sso_autologin")
+        except Exception as e:
+            log.error("Error getting auth url: {}".format(e))
+            return tk.abort(500, "Error getting auth url: {}".format(e))
+        return tk.redirect_to(auth_url + '&prompt=none')  # &prompt=none is for skipping the UI of keycloak
+
+
+# Enable this for autologin - only for / endpoint for now
+@keycloak.before_app_request
+def before_app_request():
+    # if already logged in
+    user_ckan = tk.current_user.name
+    if user_ckan:
+        pass
+    # if not logged in, check if keycloak has cookie already
+    else:
+        if tk.request.endpoint == 'home.index':
+            return tk.redirect_to('keycloak.sso_autologin')
+
+
 keycloak.add_url_rule('/sso', view_func=sso)
+keycloak.add_url_rule('/sso_autologin', view_func=sso_autologin)
 keycloak.add_url_rule('/sso_login', view_func=sso_login)
 keycloak.add_url_rule('/reset_password', view_func=reset_password, methods=['POST'])
 
+
 def get_blueprint():
-    return keycloak
+    return keycloak