import mysql.connector as mariadb import datetime from bottle import request, abort, redirect from config import database, globalvar from scripts import loggorilla, tokenguard class validation(): def __init__(self): pass def account(self, APIADDR, allowed_roles): response = {} 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) loggorilla.prcss(APIADDR, "Get dependency data: Session") cursor.execute(f"SELECT * FROM auth_session WHERE id = %s ; ", (payload["session"]["id"],) ) r_session = cursor.fetchone() loggorilla.prcss(APIADDR, "Get dependency data: Profile") 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() loggorilla.prcss(APIADDR, "Get dependency data: Roles: execute") cursor.execute(f"SELECT auth_roles FROM auth_profile_roles WHERE auth_profile = %s ; ", (r_profile['id'],) ) loggorilla.prcss(APIADDR, "Get dependency data: Roles: fetchall") r_roles = cursor.fetchall() loggorilla.fyinf(APIADDR, f"r_roles: {r_roles}") loggorilla.prcss(APIADDR, "Get dependency data: Roles: variable replace") r_profile['roles'] = [item['auth_roles'] for item in r_roles] loggorilla.prcss(APIADDR, "Get dependency data: Close DB") 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"Session ID: {r_session['id']}" ) 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"Session ID: {r_session['id']}" ) 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 ID : {r_session['id']}" ) loggorilla.accss(APIADDR, f"Session Start : {r_session['start'].strftime('%Y-%m-%d %H:%M:%S')}" ) loggorilla.accss(APIADDR, f"Session End : {r_session['end'].strftime('%Y-%m-%d %H:%M:%S')}" ) loggorilla.accss(APIADDR, f"Profile ID : {r_profile['id']}" ) loggorilla.accss(APIADDR, f"Profile Username : {r_profile['username']}" ) loggorilla.accss(APIADDR, f"Profile Email : {r_profile['email']}" ) loggorilla.accss(APIADDR, f"Profile Phone : {r_profile['phone']}" ) 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