surtur
927bdd03ce
* authorization [WIP] * refactored several sections * used dict(exclude_unset=True) * global values are now sourced from a config file (statuspagerc)
43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
import bcrypt
|
|
import jwt
|
|
from fastapi import HTTPException
|
|
from datetime import datetime, timedelta
|
|
from app import crud
|
|
from app.database import SessionLocal
|
|
from app.settings import globals as settings
|
|
|
|
|
|
|
|
KEY = settings.JWT_SECRET
|
|
ALGORITHM = settings.JWT_ALGORITHM
|
|
EXPIRY = settings.JWT_EXPIRY
|
|
|
|
|
|
def login(username: str, password: str):
|
|
db = SessionLocal()
|
|
usr = crud.login_info(db=db, name=username,password=password)
|
|
if usr is None:
|
|
return {"status": "error", "mesage": "username/password incorrect"}
|
|
description = 'access_token'
|
|
token = jwt.encode({
|
|
'sub': username,
|
|
'iat': datetime.utcnow(),
|
|
'exp': datetime.utcnow() + timedelta(seconds=EXPIRY),
|
|
'des': description
|
|
},
|
|
KEY, ALGORITHM)
|
|
return {"status": "success", "token": token.decode('utf-8')}
|
|
|
|
def validate(token):
|
|
try:
|
|
data = jwt.decode(token, KEY)
|
|
except Exception as e:
|
|
if "expired" in str(e):
|
|
raise HTTPException(status_code=401, detail={"status": "error", "message": "Token expired"})
|
|
elif "Not enough segments" in str(e):
|
|
raise HTTPException(status_code=401, detail={"status": "error", "message": "Invalid token"})
|
|
else:
|
|
raise HTTPException(status_code=400, detail={"status": "error", "message": "Exception: " + str(e)})
|
|
return data
|
|
|