123 lines
3.5 KiB
Python
123 lines
3.5 KiB
Python
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
|
|
from models import Tokens
|
|
|
|
|
|
#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')
|
|
|
|
session = db.session
|
|
|
|
|
|
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()
|
|
new_item = Tokens(tokens=hash, time=time, end_date=end_date)
|
|
session.add(new_item)
|
|
session.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()
|
|
valid = False
|
|
hashes = [row[0] for row in session.query(Tokens.tokens).all()]
|
|
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)
|
|
return valid
|
|
|
|
|
|
def check_validaty():
|
|
end_dates = [row[0] for row in session.query(Tokens.end_date).all()]
|
|
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:
|
|
session.query(Tokens).filter_by(end_date=end_date).delete()
|
|
session.commit()
|
|
|
|
|
|
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")
|
|
new_item = Tokens(tokens=hash, time=time, end_date=end_date)
|
|
session.add(new_item)
|
|
session.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()
|
|
hashes = [row[0] for row in session.query(Tokens.tokens).all()]
|
|
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()
|
|
hashes = [row[0] for row in session.query(Tokens.tokens).all()]
|
|
for hash in hashes:
|
|
try:
|
|
valid = ph.verify(hash, token)
|
|
if valid == True:
|
|
session.query(Tokens).filter_by(tokens=hash).delete()
|
|
db.connection.commit()
|
|
logging.critical("token deleted")
|
|
except Exception as e:
|
|
logging.error(e)
|