I am using flask_login, It worked in local but when i moved to different server it is did not work. On trying to debug i realized login_manager.user_loader is not getting on remote machine but the same worked in my local.
This is issue is specific to okta sso authentication on remote machine only. Post okta authentication, i am calling below to save data in session.
login_user(user)
Post this, i was expecting user_loaded to be called but that is not happening. Tried going through document but could not understand. Can you anyone help me as why user_loader is getting called in my local machine but not on remote machine.
init.py
import os
from flask import Flask
from flask_login import LoginManager
from config import app_config
login_manager = LoginManager()
def create_app(config_name):
app = Flask(__name__,
instance_path=os.path.join(os.path.abspath(os.curdir), 'instance'),
instance_relative_config=True)
app.config.from_object(app_config[config_name])
app.config.from_pyfile('config.py')
login_manager.init_app(app)
login_manager.login_message = 'You must be logged in to access this page'
login_manager.login_view = 'admin.login'
login_manager.login_message_category = 'info'
login_manager.session_protection = "strong"
#login_manager.refresh_view ='admin.login'
#login_manager.needs_refresh_message = (u"To protect your account, please reauthenticate to access this page.")
from .home import home as home_blueprint
app.register_blueprint(home_blueprint)
from .restapi import restapi as restapi_blueprint
app.register_blueprint(restapi_blueprint, url_prefix='/steps/api/v1')
return app
In my blueprint view, i am calling login_user to set current user. I have placed user_loader in one seperate file.
from flask import current_app
from flask_login import UserMixin
from app import login_manager, get_connection
from .encryption import encrypt_json, encrypt_password, decrypt_password # noqa
class User(UserMixin):
"""Custom User class."""
id = None
name = None
email = None
description = None
role = None
def __init__(self, id_, name, email, description, role):
print(name)
self.id = str(id_)
self.name = name
self.email = email
self.description = description
self.role = role
def __repr__(self):
return self.name
def claims(self):
"""Use this method to render all assigned claims on profile page."""
return {'name': self.name,
'email': self.email}.items()
@staticmethod
def get(user_id):
try:
connection = get_connection()
user_sql = "SELECT ID, USER_NAME, DESCRIPTION, EMAIL, ROLE FROM USER WHERE ID = {}"
cursor = connection.cursor()
cursor.execute(user_sql.format(user_id))
data = cursor.fetchone()
if data:
return User(id_=data[0], name=data[1], email=data[3], description=data[2], role=data[4])
else:
return None
except Exception as ex:
print(ex)
finally:
cursor.close()
connection.close()
# print(user_id)
# print(USERS_DB)
# return USERS_DB.get(user_id)
@staticmethod
def getByName(user_name):
try:
connection = get_connection()
user_sql = "SELECT ID, USER_NAME, DESCRIPTION, EMAIL, ROLE FROM USER WHERE USER_NAME = '{}' "
cursor = connection.cursor()
cursor.execute(user_sql.format(user_name))
data = cursor.fetchone()
if data:
return User(id_=data[0], name=data[1], email=data[3], description=data[2], role=data[4])
else:
return None
except Exception as ex:
print(ex)
finally:
cursor.close()
connection.close()
@staticmethod
def getByEmail(emailid):
try:
connection = get_connection()
user_sql = "SELECT ID, USER_NAME, DESCRIPTION, EMAIL, ROLE FROM USER WHERE EMAIL = '{}' "
cursor = connection.cursor(buffered=True)
cursor.execute(user_sql.format(emailid))
data = cursor.fetchone()
print(data)
if data:
return User(id_=data[0], name=data[1], email=data[3], description=data[2], role=data[4])
else:
return None
except Exception as ex:
print(ex)
finally:
cursor.close()
connection.close()
@staticmethod
def verify_password(username, password):
try:
# mysql_hook = MySqlHook(mysql_conn_id="STEPS", schema="STEPS")
# connection = mysql_hook.get_conn()
connection = get_connection()
cursor = connection.cursor()
encFlag, decryptedPassword = decrypt_password(password)
lognReqTuple = (username, password)
user_sql = "SELECT USER_NAME, PASSWORD from USER WHERE USER_NAME = '{}'"
cursor.execute(user_sql.format(username))
all_users = cursor.fetchall()
print(all_users)
for user in all_users:
encFlag, decryptedPassword = decrypt_password(user[1]) if user[1] else (False, password)
existingUserTuple = user[0], decryptedPassword
if existingUserTuple == lognReqTuple:
# logging.info('User {} authenticated.'.format(username))
return 0
# logging.info('Un Authorized access, user authentication failed.')
return 1
except Exception as ex:
# logging.error('Error in verifying login details :{}'.format(ex))
print(ex)
return 1
finally:
cursor.close()
connection.close()
@staticmethod
def create(user_id, name, email):
# USERS_DB[user_id] = User(user_id, name, email)
pass
@login_manager.user_loader
def user_loader(user_id):
print('user loader', type(user_id), user_id)
return User.get(user_id)
Problem is that, same setup is working in my local ubuntu environment but when i moved to production vm on centos, it works for one view where i am using local authentication but not in case of okta.
Not sure about the issue but i changed the flask app folder structure. changed the way i was using blueprints and issue got solved. All in all, had to change complete application structure.