Stanislav Hejny
59 lines
1.5 KiB
Python
59 lines
1.5 KiB
Python
from pytz import timezone
|
|
|
|
from typing import Optional
|
|
from datetime import datetime, timedelta
|
|
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
|
|
import jwt
|
|
|
|
from core.configs import settings
|
|
from core.security import validate_pass, create_password_hash
|
|
from schemas.user_schema import UserSchema
|
|
|
|
oauth2_schema = OAuth2PasswordBearer(
|
|
tokenUrl=f"{settings.API_V0_STR}/login"
|
|
)
|
|
|
|
|
|
async def authenticate(username: str, password: str) -> Optional[UserSchema]:
|
|
if username != 'user':
|
|
return None
|
|
|
|
test_hash = create_password_hash('user1234')
|
|
|
|
if not validate_pass(password, test_hash):
|
|
return None
|
|
|
|
# The returned value at this point should be UserModel, but since we do not have any persistence
|
|
# and this is for demo purposes only
|
|
return UserSchema(id=0, username='user', hash_pass=test_hash)
|
|
|
|
def _create_token(token_type: str, lifetime: timedelta, sub: str) -> str:
|
|
# https://dataracker.ietf.org/doc/html/rfc7519#section-4.1.3
|
|
payload = {}
|
|
|
|
pt = timezone('UTC')
|
|
expires = datetime.now(tz=pt) + lifetime
|
|
|
|
payload["type"] = token_type
|
|
|
|
payload["exp"] = expires
|
|
|
|
payload["iat"] = datetime.now(tz=pt) #Issued-At (iat)
|
|
|
|
payload["sub"] = str(sub)
|
|
|
|
return jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.ALGORITHM)
|
|
|
|
|
|
def create_access_token(sub: str) -> str:
|
|
"""
|
|
https://jwt.io
|
|
"""
|
|
return _create_token(
|
|
token_type='access_token',
|
|
lifetime=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES),
|
|
sub=sub
|
|
)
|