import mysql.connector as mariadb import datetime from bottle import request from config import database, globalvar from scripts import loggorilla, tokenguard class validation(): def __init__(self): pass def account(self, APIADDR, allowed_roles): response = {} try: loggorilla.prcss(APIADDR, "Define parameters") beaker_session = request.environ.get('beaker.session') jwt = beaker_session["token"] if "token" in beaker_session else None if jwt is None: loggorilla.fyinf(APIADDR, "Guest") r_session = {} r_profile = { "username" :None, "email" :None, "phone" :None, "roles" :[0] } else: loggorilla.fyinf(APIADDR, "With JWT") loggorilla.prcss(APIADDR, "Get JWT payload data") payload = tokenguard.decode(jwt, globalvar.ssh['key']['public']) loggorilla.prcss(APIADDR, "Get dependency data") db_main = mariadb.connect(**database.db_main) cursor = db_main.cursor(dictionary=True) cursor.execute(f"SELECT * FROM auth_session WHERE id = %s ; ", (payload["session"]["id"],) ) r_session = cursor.fetchone() cursor.execute(f"SELECT COUNT(*) AS `count`, auth_profile.* FROM auth_profile_verification LEFT JOIN auth_profile ON auth_profile.id = auth_profile_verification.auth_profile WHERE auth_profile.token = %s AND auth_profile_verification.type = 'email' AND auth_profile_verification.verified = 1 ; ", (r_session['token'],) ) r_profile = cursor.fetchone() cursor.execute(f"SELECT auth_roles FROM auth_profile_roles WHERE auth_profile = %s ; ", (r_profile['id'],) ) r_roles = cursor.fetchall() r_profile['roles'] = r_roles['auth_roles'] cursor.close() db_main.close() loggorilla.prcss(APIADDR, "Validation") if 0 not in r_profile['roles'] and datetime.datetime.now() > r_session['end']: loggorilla.prcss(APIADDR, "Deleting") self.cursor.execute("DELETE FROM auth_session WHERE id = %s ; ", (r_session['id'],) ) loggorilla.prcss(APIADDR, "Giving response") loggorilla.accss(APIADDR, "Expired. Your session removed." ) loggorilla.accss(APIADDR, f"Session: {r_session}" ) loggorilla.accss(APIADDR, f"Profile: {r_profile}" ) response["status" ] = "failed" response["desc" ] = "Expired. Your session removed." response["data" ] = { "valid" :{ "status" : 0, "desc" : "expired" }, "session" : r_session, "profile" : r_profile } redirect('/logout?msg=expired') elif 0 not in r_profile['roles'] and r_profile["count"] == 0: loggorilla.prcss(APIADDR, "Giving response") loggorilla.accss(APIADDR, "No active account for this" ) loggorilla.accss(APIADDR, f"Session: {r_session}" ) loggorilla.accss(APIADDR, f"Profile: {r_profile}" ) response["status" ] = "failed" response["desc" ] = "No active account for this" response["data" ] = { "message" : "Please contact us if you still had a problem", "valid" :{ "status" : 0, "desc" : "fake" }, "session" : r_session, "profile" : r_profile } abort(403, "Please contact us if you still had a problem.") # 403 Forbidden elif any(role in allowed_roles for role in r_profile['roles']): loggorilla.prcss(APIADDR, "Giving response") loggorilla.accss(APIADDR, "User roles authorized" ) response["status" ] = "success" response["desc" ] = "User roles authorized" response["data" ] = { "valid" :{ "status" : 1, "desc" : "authorized" }, "session" : r_session, "profile" : r_profile } return response else: loggorilla.prcss(APIADDR, "Giving response") loggorilla.accss(APIADDR, "User roles unauthorized" ) loggorilla.accss(APIADDR, f"Session: {r_session}" ) loggorilla.accss(APIADDR, f"Profile: {r_profile}" ) response["status" ] = "failed" response["desc" ] = "User roles unauthorized" response["data" ] = { "valid" :{ "status" : 0, "desc" : "unauthorized" }, "session" : r_session, "profile" : r_profile } abort(401, "User roles unauthorized") # 401 Unauthorized except Exception as e: loggorilla.error(APIADDR, str(e) ) response["status" ] = "failed" response["desc" ] = "Internal Server Error. Please contact us if you still have an error." return response