95 lines
2.7 KiB
Python
95 lines
2.7 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
from typing import Annotated
|
|
|
|
import jwt
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from pwdlib import PasswordHash
|
|
|
|
from app.config import Settings
|
|
from app.database import get_from_database_where
|
|
from app.models.token import TokenData
|
|
from app.models.user import User, UserInDB
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
|
|
|
password_hash = PasswordHash.recommended()
|
|
|
|
DUMMY_HASH = password_hash.hash("testpassword")
|
|
|
|
|
|
def verify_password(plain_password: str, hashed_password: str):
|
|
return password_hash.verify(plain_password, hashed_password)
|
|
|
|
|
|
def get_password_hash(password: str):
|
|
return password_hash.hash(password)
|
|
|
|
|
|
def get_user(username: str):
|
|
user = get_from_database_where(
|
|
User,
|
|
User.username, # type: ignore (Type correct on runtime)
|
|
username,
|
|
True,
|
|
)
|
|
if user is not None:
|
|
return UserInDB.model_validate(user[0])
|
|
|
|
|
|
def authenticate_user(username: str, password: str):
|
|
user = get_user(username)
|
|
if not user:
|
|
verify_password(password, DUMMY_HASH)
|
|
return False
|
|
if not verify_password(password, user.hashed_password):
|
|
return False
|
|
return user
|
|
|
|
|
|
def create_access_token(
|
|
data: dict[str, str | datetime],
|
|
expires_delta: timedelta | None = None,
|
|
):
|
|
to_encode = data.copy()
|
|
if expires_delta:
|
|
expire = datetime.now(UTC) + expires_delta
|
|
else:
|
|
expire = datetime.now(UTC) + timedelta(minutes=15)
|
|
to_encode.update({"exp": expire})
|
|
encoded_jwt = jwt.encode( # pyright: ignore[reportUnknownMemberType] (Third party skill issue)
|
|
to_encode,
|
|
Settings.secret_key,
|
|
algorithm=Settings.algorithm,
|
|
)
|
|
return encoded_jwt
|
|
|
|
|
|
async def get_current_user(token: Annotated[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( # pyright: ignore[reportUnknownMemberType] (Third party skill issue)
|
|
token,
|
|
Settings.secret_key,
|
|
algorithms=[Settings.algorithm],
|
|
)
|
|
username = payload.get("sub")
|
|
if username is None:
|
|
raise credentials_exception
|
|
token_data = TokenData(username=username)
|
|
except jwt.InvalidTokenError:
|
|
raise credentials_exception
|
|
user = get_user(token_data.username) # type: ignore Look just above here idiot!
|
|
if user is None:
|
|
raise credentials_exception
|
|
return user
|
|
|
|
|
|
async def get_current_active_user(
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
):
|
|
return current_user
|