import time from dotenv import load_dotenv import db import os from datetime import datetime, timedelta import logging from argon2 import PasswordHasher import random import string #Logging logging.basicConfig( filename='api.log', # Log to a file level=logging.DEBUG, # Log everything (DEBUG and above) format='%(asctime)s %(levelname)s %(message)s' ) logging.info('Application started') try: logging.debug('Debug info here') except Exception as e: logging.exception('An error occurred') load_dotenv() db_name = os.getenv("DB_NAME") db_psswd = os.getenv("DB_PASSWORD") db_user = os.getenv("DB_USER") db_port = os.getenv("DB_PORT") or "3306" db_host = os.getenv("DB_HOST") if db_host == None: logging.error("no hostname provided") os._exit(1) db.connect(db_user, db_psswd, db_host, int(db_port), db_name) def generate_token(duration): ph = PasswordHasher() token = "".join(random.choices(string.ascii_lowercase + string.digits, k=32)) print(token) hash = ph.hash(token) end_date = enddate(duration) time = datetime.now() cursor = db.connection.cursor() executed = "INSERT INTO tokens (tokens, time, end_date) VALUES (?, ?, ?)" executed_data = (hash, time, end_date) cursor.execute(executed, executed_data) db.connection.commit() logging.info(f"Token till {end_date} created") return token def enddate(duration): if duration == "week": return datetime.now() + timedelta(days=7) if duration == "day": return datetime.now() + timedelta(days=1) if duration == "year": return datetime.now() + timedelta(days=365) if isinstance(duration, int): return datetime.now() + timedelta(days=duration) if duration == "forever": return datetime.now() + timedelta(days=4000) raise ValueError(f"Invalid duration: {duration}") def check_token(token): ph = PasswordHasher() cursor = db.connection.cursor() valid = False cursor.execute("SELECT tokens FROM tokens") hashes = [row[0] for row in cursor.fetchall()] for hash in hashes: try: valid = ph.verify(hash, token) if valid == True: return valid except Exception as e: logging.error("Token not the same") logging.error(e) def check_validaty(): cursor = db.connection.cursor() cursor.execute("SELECT end_date FROM tokens") end_dates = [row[0] for row in cursor.fetchall()] for end_date in end_dates: end_date_dt = datetime.strptime(end_date, "%Y-%m-%d %H:%M:%S") if datetime.now() > end_date_dt: cursor.execute("DELETE FROM tokens WHERE end_date = ?", (end_date,)) def admin_token(): ph = PasswordHasher() token = "".join(random.choices(string.ascii_lowercase + string.digits, k=64)) hash = ph.hash(token) time = datetime.now() end_date = enddate("forever") cursor = db.connection.cursor() executed = "INSERT INTO tokens (tokens, time, end_date) VALUES (?, ?, ?)" executed_data = (hash, time, end_date) cursor.execute(executed, executed_data) db.connection.commit() logging.info(f"Token till {end_date} created") return token def check_admin_token(token): valid = False if len(token) == 64: ph = PasswordHasher() cursor = db.connection.cursor() cursor.execute("SELECT tokens FROM tokens") hashes = [row[0] for row in cursor.fetchall()] for hash in hashes: try: valid = ph.verify(hash, token) if valid == True: return valid except Exception as e: logging.info(e) def delete_token(token): if not len(token) == 64: ph = PasswordHasher() cursor = db.connection.cursor() cursor.execute("SELECT tokens FROM tokens") hashes = [row[0] for row in cursor.fetchall()] for hash in hashes: try: valid = ph.verify(hash, token) if valid == True: cursor.execute("DELETE FROM tokens WHERE tokens = ?", (hash,)) db.connection.commit() logging.critical("token deleted") except Exception as e: logging.error(e)