API/app/auth.py

108 lines
3.5 KiB
Python
Raw Normal View History

2021-12-03 12:53:55 +00:00
from os import getenv
import json
from datetime import timedelta, datetime
from typing import Optional
from fastapi.security import OAuth2PasswordBearer
2022-01-12 07:48:18 +00:00
from fastapi import Depends, HTTPException, status, APIRouter
2021-12-03 12:53:55 +00:00
from jose import JWTError, jwt
from passlib.context import CryptContext
2022-01-12 07:48:18 +00:00
from app.models import User, UserInDB, TokenData, Token
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
2021-12-03 12:53:55 +00:00
2022-01-12 07:48:18 +00:00
router = APIRouter()
2021-12-03 12:53:55 +00:00
# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = getenv("SECRET_KEY")
ALGORITHM = getenv("ALGORITHM")
ACCESS_TOKEN_EXPIRE_MINUTES = int(getenv("ACCESS_TOKEN_EXPIRE_MINUTES"))
fake_users_db = json.load(open("app/users.json"))
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return pwd_context.hash(password)
def get_user(db, username: str):
if username in db:
user_dict = db[username]
return UserInDB(**user_dict)
def authenticate_user(fake_db, username: str, password: str):
user = get_user(fake_db, username)
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return user
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=token_data.username)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(current_user: User = Depends(get_current_user)):
if current_user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
2022-01-12 07:48:18 +00:00
return current_user
def authorise(server, current_user: User = Depends(get_current_active_user)):
print(server, current_user.servers)
if server not in current_user.servers:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
@router.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}