CHIEFSOFT\ameye
135 lines
5.8 KiB
Python
135 lines
5.8 KiB
Python
from flask import session, jsonify
|
|
from marshmallow import ValidationError
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
import datetime
|
|
import jwt
|
|
from app.config import Config
|
|
from app.models import OfficeUsers
|
|
from app.utils.logger import logger
|
|
from app.extensions import db
|
|
|
|
class OfficeAuthService:
|
|
@staticmethod
|
|
def login(username, password):
|
|
"""
|
|
Login method that checks for specific credentials and returns a JWT token
|
|
"""
|
|
# Define valid credentials for testing
|
|
valid_credentials = {
|
|
"mermsuser": "mermsuser",
|
|
"admin": "admin123",
|
|
"test": "test123"
|
|
}
|
|
|
|
logger.info('ENTER API:: login')
|
|
try:
|
|
with db.session.begin():
|
|
member = OfficeUsers.get_office_user_by_username(username)
|
|
password_hash = generate_password_hash(password)
|
|
logger.info("Password generated = > {}".format(password_hash))
|
|
|
|
if not member:
|
|
invalid_data = {
|
|
"error_message": "invalid username or password",
|
|
"message_key": "invalid_username_or_password",
|
|
}
|
|
return invalid_data, 401
|
|
user_id = member.id
|
|
member_password = member.password
|
|
logger.info("Current Password = > {}".format(member_password))
|
|
if str(member_password).strip() == 'password':
|
|
updateResult = OfficeUsers.set_office_user_password(user_id, username, password_hash)
|
|
logger.info(f"Password Update Result = > {updateResult} ")
|
|
member_password= password_hash
|
|
member = OfficeUsers.get_office_user_by_username(username) # reload office
|
|
|
|
pass_check = check_password_hash(member.password, password)
|
|
logger.info("Password check: {}".format(pass_check))
|
|
if not member or not pass_check:
|
|
invalid_data = {
|
|
"error_message": "invalid username or password",
|
|
"message_key": "invalid_username_or_password",
|
|
}
|
|
return invalid_data , 401
|
|
else:
|
|
payload = {
|
|
'sub': username, # Subject (typically user ID)
|
|
'iat': datetime.datetime.utcnow(), # Issued at
|
|
'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=15), # Expiration (15 minutes)
|
|
'role': 'admin' if username == 'admin' else 'user' # Role based on username
|
|
}
|
|
|
|
# Get the secret key from config
|
|
secret_key = Config.JWT_SECRET_KEY
|
|
|
|
# Generate the token
|
|
token = jwt.encode(payload, secret_key, algorithm='HS256')
|
|
|
|
# Return the token and user info
|
|
return {
|
|
'jwt_token': token,
|
|
'user': {
|
|
'firstname': member.firstname,
|
|
'lastname': member.lastname,
|
|
'email':'support@mermsemr.com',
|
|
'username': username,
|
|
'role': 'admin' if username == 'admin' else 'user'
|
|
},
|
|
'expires_in': 900 # 15 minutes in seconds
|
|
}
|
|
|
|
|
|
|
|
# # Check if the provided credentials are valid
|
|
# if username in valid_credentials and password == valid_credentials[username]:
|
|
# # Generate JWT token with 15 minutes expiration
|
|
# payload = {
|
|
# 'sub': username, # Subject (typically user ID)
|
|
# 'iat': datetime.datetime.utcnow(), # Issued at
|
|
# 'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=15), # Expiration (15 minutes)
|
|
# 'role': 'admin' if username == 'admin' else 'user' # Role based on username
|
|
# }
|
|
#
|
|
# # Get the secret key from config
|
|
# secret_key = Config.JWT_SECRET_KEY
|
|
#
|
|
# # Generate the token
|
|
# token = jwt.encode(payload, secret_key, algorithm='HS256')
|
|
#
|
|
# # Return the token and user info
|
|
# return {
|
|
# 'jwt_token': token,
|
|
# 'user': {
|
|
# 'username': username,
|
|
# 'role': 'admin' if username == 'admin' else 'user'
|
|
# },
|
|
# 'expires_in': 900 # 15 minutes in seconds
|
|
# }
|
|
# else:
|
|
# # Return error for invalid credentials
|
|
# return {
|
|
# 'error': 'Invalid credentials',
|
|
# 'message': 'The username or password is incorrect'
|
|
# }, 401
|
|
|
|
except Exception as e:
|
|
logger.error(f"An error occurred while get_office_country_list data: {str(e)}", exc_info=True)
|
|
return jsonify({"message": "Internal Server Error"}), 500
|
|
|
|
|
|
@staticmethod
|
|
def verify_token(token):
|
|
"""
|
|
Verify the JWT token
|
|
"""
|
|
try:
|
|
# Get the secret key from config
|
|
secret_key = Config.JWT_SECRET_KEY
|
|
|
|
# Decode the token
|
|
payload = jwt.decode(token, secret_key, algorithms=['HS256'])
|
|
return payload
|
|
except jwt.ExpiredSignatureError:
|
|
return None # Token has expired
|
|
except jwt.InvalidTokenError:
|
|
return None # Invalid token |